Trim preprocessor (#25)

* add "needs-trim" test file for testing trim_pose

* add spoken-to-signed to reqs for trim_pose

* add trim_pose PoseProcessor

* test trim_pose PoseProcessor

* pylint changes

* fixing pytests to deal with new test file

* run black

* Turns out detect_known_format crashes when given a pose with no components

Author: cleong110

pose_evaluation/metrics/conftest.py

@@ -57,6 +57,13 @@ def _check_range(
 
 @pytest.fixture
 def real_pose_files() -> List[Pose]:
-    test_files_folder = Path("pose_evaluation") / "utils" / "test" / "test_data"
+    # pose_evaluation/utils/test/test_data/standard_landmarks
+    test_files_folder = Path("pose_evaluation") / "utils" / "test" / "test_data" / "mediapipe" / "standard_landmarks"
     real_pose_files_list = [Pose.read(test_file.read_bytes()) for test_file in test_files_folder.glob("*.pose")]
     return real_pose_files_list
+
+
+@pytest.fixture
+def real_refined_landmark_pose_file_paths() -> List[Path]:
+    test_files_folder = Path("pose_evaluation") / "utils" / "test" / "test_data" / "mediapipe" / "refined_landmarks"
+    return list(test_files_folder.glob("*.pose"))

pose_evaluation/metrics/pose_processors.py

@@ -3,13 +3,16 @@
 from tqdm import tqdm
 
 from pose_format import Pose
-from pose_format.utils.generic import pose_hide_legs, reduce_holistic
+from pose_format.utils.generic import pose_hide_legs, reduce_holistic, detect_known_pose_format
+from spoken_to_signed.gloss_to_pose.concatenate import trim_pose
+
 from pose_evaluation.metrics.base import Signature
 from pose_evaluation.utils.pose_utils import (
     zero_pad_shorter_poses,
     reduce_poses_to_intersection,
 )
 
+
 PosesTransformerFunctionType = Callable[[Iterable[Pose]], List[Pose]]
 
 
@@ -117,7 +120,22 @@ def process_pose(self, pose: Pose) -> Pose:
         return pose
 
 
+class TrimMeaninglessFramesPoseProcessor(PoseProcessor):
+    def __init__(self, start=True, end=True) -> None:
+        super().__init__(name="trim_pose")
+        self.start = start
+        self.end = end
+
+    def process_pose(self, pose):
+        if detect_known_pose_format(pose) == "holistic":
+
+            return trim_pose(pose.copy(), start=self.start, end=self.end)
+        # not supported
+        return pose
+
+
 def get_standard_pose_processors(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    trim_meaningless_frames: bool = True,
     normalize_poses: bool = True,
     reduce_poses_to_common_components: bool = True,
     remove_world_landmarks=True,
@@ -128,6 +146,12 @@ def get_standard_pose_processors(  # pylint: disable=too-many-arguments,too-many
 ) -> List[PoseProcessor]:
     pose_processors = []
 
+    # remove leading/trailing frames with no hands in frame.
+    if trim_meaningless_frames:
+        pose_processors.append(TrimMeaninglessFramesPoseProcessor())
+
+    # Note: by default this uses the shoulder joints,
+    # so it should be BEFORE anything that might remove those, such as reduce poses to common components
     if normalize_poses:
         pose_processors.append(NormalizePosesProcessor())
 

pose_evaluation/metrics/test_distance_metric.py

@@ -70,6 +70,7 @@ def setUp(self):
             distance_measure=AggregatedPowerDistance(order=1, default_distance=0),
             # preprocessors that won't crash
             pose_preprocessors=get_standard_pose_processors(
+                trim_meaningless_frames=False,  # fake poses have no components, this crashes.
                 normalize_poses=False,
                 remove_world_landmarks=False,
                 remove_legs=False,
@@ -99,6 +100,7 @@ def setUp(self):
             distance_measure=AggregatedPowerDistance(order=2, default_distance=self.default_distance),
             # preprocessors that won't crash
             pose_preprocessors=get_standard_pose_processors(
+                trim_meaningless_frames=False,  # fake poses don't have the necessary components
                 normalize_poses=False,
                 remove_world_landmarks=False,
                 remove_legs=False,

pose_evaluation/metrics/test_dtw_metric.py

@@ -12,6 +12,7 @@ def setUp(self):
             name="DTWPowerDistance",
             distance_measure=distance_measure,
             pose_preprocessors=get_standard_pose_processors(
+                trim_meaningless_frames=False,  # fake poses have no components, this crashes.
                 normalize_poses=False,  # no shoulders, will crash
                 remove_world_landmarks=False,  # there are none, will crash
                 reduce_poses_to_common_components=False,  # removes all components, there are none in common

pose_evaluation/metrics/test_pose_processors.py

@@ -0,0 +1,28 @@
+from typing import List
+from pathlib import Path
+from pose_format import Pose
+
+from pose_evaluation.metrics.pose_processors import TrimMeaninglessFramesPoseProcessor
+
+
+def test_trim_pose(real_refined_landmark_pose_file_paths: List[Path]):
+
+    for pose_path in real_refined_landmark_pose_file_paths:
+        if "needs-trim" in pose_path.name:
+            pose = Pose.read(pose_path.read_bytes())
+
+            original_frame_count = pose.body.data.shape[0]
+
+            processor = TrimMeaninglessFramesPoseProcessor(start=True, end=True)
+
+            processed_pose = processor.process_pose(pose)
+
+            # not expecting it to edit the original
+            assert (
+                pose.body.data.shape[0] == original_frame_count
+            ), f"Original data changed! Frames before: {original_frame_count}. Now: {pose.body.data.shape[0]}"
+
+            # should have fewer frames
+            assert (
+                processed_pose.body.data.shape[0] < pose.body.data.shape[0]
+            ), f"{pose_path}, {pose.body}, {processed_pose.body}"

pose_evaluation/utils/conftest.py

@@ -13,12 +13,14 @@
 from pose_evaluation.utils.pose_utils import load_pose_file
 
 
-utils_test_data_dir = Path(__file__).parent / "test" / "test_data"
+utils_standard_mediapipe_landmarks_test_data_dir = (
+    Path(__file__).parent / "test" / "test_data" / "mediapipe" / "standard_landmarks"
+)
 
 
 @pytest.fixture(scope="function")
 def mediapipe_poses_test_data_paths() -> List[Path]:
-    pose_file_paths = list(utils_test_data_dir.glob("*.pose"))
+    pose_file_paths = list(utils_standard_mediapipe_landmarks_test_data_dir.glob("*.pose"))
     return pose_file_paths
 
 
@@ -33,7 +35,7 @@ def mediapipe_poses_test_data(mediapipe_poses_test_data_paths) -> List[Pose]:  #
 
 @pytest.fixture
 def standard_mediapipe_components_dict() -> Dict[str, List[str]]:
-    format_json = utils_test_data_dir / "mediapipe_components_and_points.json"
+    format_json = utils_standard_mediapipe_landmarks_test_data_dir / "mediapipe_components_and_points.json"
     with open(format_json, "r", encoding="utf-8") as f:
         return json.load(f)
 

pose_evaluation/utils/test/test_data/mediapipe/refined_landmarks/colin-HOUSE-needs-trim.pose

@@ -22,6 +22,8 @@ dependencies = [
     "fastdtw",
     # alternative to fastdtw
     "dtaidistance",
+    # so that we can have the "trim_pose" preprocessor
+    "spoken-to-signed @ git+https://github.com/ZurichNLP/spoken-to-signed-translation.git",
 ]
 
 [project.optional-dependencies]