add nwb table for lightning pose face parts results

src/npc_sessions/sessions.py

@@ -2716,6 +2716,48 @@ def _video_frame_times(
             for path, timestamps in path_to_timestamps.items()
         )
 
+    @npc_io.cached_property
+    def _LPFaceParts(self) -> tuple[pynwb.core.DynamicTable, ...]:
+        """
+        Stores the lightning pose output as a dynamic table for each of the relevant cameras (side, face)
+        """
+        LP_face_parts_dynamic_tables = []
+        if not self.is_video:
+            raise ValueError(f"{self.id} is not a session with video")
+
+        for video_path in self.video_paths:
+            camera_name = npc_mvr.get_camera_name(video_path.name)
+            if camera_name == "eye":
+                continue
+
+            nwb_camera_name = self.mvr_to_nwb_camera_name[camera_name]
+            timestamps = next(t for t in self._video_frame_times if nwb_camera_name in t.name).timestamps
+            assert len(timestamps) == npc_mvr.get_total_frames_in_video(video_path)
+            df = utils.get_LPFaceParts_predictions_dataframe(self.id, utils.LP_MAPPING[camera_name])
+            if len(timestamps) != len(df):
+                logger.warning(
+                    f"{self.id} {camera_name} lightning pose face parts output has wrong shape {len(df)}, expected {len(timestamps)} frames."
+                    "\nLightning pose face parts capsule was likely run with an additional data asset attached"
+                )
+                continue
+
+            df['timestamps'] = timestamps
+            name = f"Lightning_Pose_FaceParts_{nwb_camera_name}"
+            table_description = (
+                f"Lightning Pose tracking model fit to {len(utils.LP_VIDEO_FEATURES_MAPPING[utils.LP_MAPPING[camera_name]])} facial features for each frame of {nwb_camera_name} video. "
+                "Output for every frame is x,y coordinates in pixels along with the likelihood of the model for each feature in the frame. "
+                f"Features tracked are {utils.LP_VIDEO_FEATURES_MAPPING[utils.LP_MAPPING[camera_name]]} "
+            )
+            table = pynwb.core.DynamicTable.from_dataframe(
+                name=name,
+                table_description=table_description,
+                df=df
+            )
+            LP_face_parts_dynamic_tables.append(table)
+
+        return tuple(LP_face_parts_dynamic_tables)
+
+
     @npc_io.cached_property
     def _eye_tracking(self) -> pynwb.core.DynamicTable:
         if not self.is_video:

src/npc_sessions/utils/videos.py

@@ -20,7 +20,11 @@
 }
 
 FACEMAP_CAMERA_NAMES: tuple[npc_mvr.CameraName, ...] = ("behavior", "face")
-
+LP_CAMERA_NAMES = ("side", "face")
+LP_MAPPING = {"behavior": "side", "face": "face"}
+LP_VIDEO_FEATURES_MAPPING = {'side': ('eye_top_l', 'eye_bottom_l', 'whisker_pad_l_top', 'whisker_pad_l_side', 'ear_tip_l', 'ear_base_l', 'nostril_l', 'nose_tip', 'jaw', 'tongue_base_l', 'tongue_tip'),
+               'face': ('eye_top_l', 'eye_bottom_l', 'whisker_pad_l_top', 'whisker_pad_l_side', 'ear_tip_l', 'ear_base_l', 'nostril_l', 'nostril_r', 'nose_tip', 'tongue_base_l', 'tongue_tip', 'tongue_base_r')
+            }
 
 def get_dlc_session_paf_graph(session: str, model_name: str) -> list:
     """
@@ -61,14 +65,31 @@ def h5_to_dataframe(h5_path: upath.UPath, key_name: str | None = None) -> pd.Dat
 
 
 def get_LPFaceParts_predictions_dataframe(
-    session: str, camera: Literal["side", "face"]
+    session: str, camera: str
 ) -> pd.DataFrame:
     """
-    Gets the result dataframe with the lightning pose prediction for the facial features for the given camera
+    Gets the result dataframe with the lightning pose prediction for the facial features for the given camera.
+    Modifies the result dataframe for easier use
     >>> df_predictions = get_LPFaceParts_predictions_dataframe('702136_2024-03-07', 'side')
     >>> len(df_predictions.columns)
-    34
+    33
+    >>> df_predictions.columns
+    Index(['eye_top_l_x', 'eye_top_l_y', 'eye_top_l_likelihood', 'eye_bottom_l_x',
+           'eye_bottom_l_y', 'eye_bottom_l_likelihood', 'whisker_pad_l_top_x',
+           'whisker_pad_l_top_y', 'whisker_pad_l_top_likelihood',
+           'whisker_pad_l_side_x', 'whisker_pad_l_side_y',
+           'whisker_pad_l_side_likelihood', 'ear_tip_l_x', 'ear_tip_l_y',
+           'ear_tip_l_likelihood', 'ear_base_l_x', 'ear_base_l_y',
+           'ear_base_l_likelihood', 'nostril_l_x', 'nostril_l_y',
+           'nostril_l_likelihood', 'nose_tip_x', 'nose_tip_y',
+           'nose_tip_likelihood', 'jaw_x', 'jaw_y', 'jaw_likelihood',
+           'tongue_base_l_x', 'tongue_base_l_y', 'tongue_base_l_likelihood',
+           'tongue_tip_x', 'tongue_tip_y', 'tongue_tip_likelihood'],
+          dtype='object')
     """
+    if camera not in LP_CAMERA_NAMES:
+        raise ValueError(f'Undefined camera name {camera}. Must be one of either side or face')
+
     session_LP_predictions_s3_paths = (
         npc_lims.get_lpfaceparts_camera_predictions_s3_paths(session, camera)
     )
@@ -84,8 +105,17 @@ def get_LPFaceParts_predictions_dataframe(
 
     df_predictions = pd.read_csv(session_LP_prediction_csv_s3_path[0], low_memory=False)
     data_array = df_predictions.to_numpy()
-    return pd.DataFrame(data_array[1:], columns=data_array[0])
-
+    df_predictions_rearranged = pd.DataFrame(data_array[1:], columns=data_array[0])
+    df_predictions_rearranged.drop(columns='bodyparts', inplace=True)
+
+    new_column_labels = []
+    for column in df_predictions_rearranged.columns.unique():
+        df_column = df_predictions_rearranged[column].iloc[0]
+        new_column_labels.append(f'{column}_{df_column.iloc[0]}')
+        new_column_labels.append(f'{column}_{df_column.iloc[1]}')
+        new_column_labels.append(f'{column}_{df_column.iloc[2]}')
+
+    return pd.DataFrame(data_array[1:, 1:], columns=new_column_labels).drop(0).reset_index(drop=True)
 
 def get_ellipse_session_dataframe_from_h5(session: str) -> pd.DataFrame:
     """