Pose distance metrics from Ham2Pose

.gitignore

@@ -1,4 +1,10 @@
 .idea/
 build/
 pose_evaluation.egg-info/
-**/__pycache__/
+**/__pycache__/
+.coverage
+.vscode/
+coverage.lcov
+**/test_data/
+*.npz
+*.code-workspace

pose_evaluation/metrics/.gitignore

@@ -1 +1 @@
-temp/
+tests

pose_evaluation/metrics/aggregate_distances.py

@@ -0,0 +1,46 @@
+from typing import Callable, Iterable, Literal
+from numpy.ma import MaskedArray
+from pose_evaluation.metrics.base import Signature, SignatureMixin
+
+
+AggregationStrategy = Literal["sum", "mean", "max"]
+DistanceAggregatorFunction = Callable[[Iterable[float]], float]
+
+
+class DistanceAggregatorSignature(Signature):
+    def __init__(self, args: dict):
+        super().__init__(args)
+        self.update_signature_and_abbr("aggregation_strategy", "s", args)
+
+
+class DistanceAggregator(SignatureMixin):
+    _SIGNATURE_TYPE = DistanceAggregatorSignature
+
+    def __init__(self, aggregation_strategy: AggregationStrategy) -> None:
+        self.aggregator_function = get_aggregator_function(
+            strategy=aggregation_strategy
+        )
+        self.aggregation_strategy = aggregation_strategy
+
+    def __call__(self, distances: Iterable[float]) -> float:
+        return self.aggregator_function(distances)
+
+
+def create_maskedarrray_and_cast_result_to_float(
+    callable_to_wrap: Callable,
+) -> DistanceAggregatorFunction:
+    return lambda a: float(callable_to_wrap(MaskedArray(a)))
+
+
+def get_aggregator_function(
+    strategy: AggregationStrategy,
+) -> DistanceAggregatorFunction:
+
+    if strategy == "max":
+        return create_maskedarrray_and_cast_result_to_float(MaskedArray.max)
+
+    if strategy == "mean":
+        return create_maskedarrray_and_cast_result_to_float(MaskedArray.mean)
+
+    if strategy == "sum":
+        return create_maskedarrray_and_cast_result_to_float(MaskedArray.sum)

pose_evaluation/metrics/base.py

@@ -1,11 +1,76 @@
 # pylint: disable=undefined-variable
 from tqdm import tqdm
+from typing import Any, Callable
 
+class Signature:
+    """Represents reproducibility signatures for metrics. Inspired by sacreBLEU
+    """
+    def __init__(self, args: dict):
+
+        self._abbreviated = {
+            "name":"n",
+            "higher_is_better":"hb"
+        }
+
+        self.signature_info = {
+            "name": args.get("name", None),
+            "higher_is_better": args.get("higher_is_better", None)
+        }
+
+    def update(self, key: str, value: Any):
+        self.signature_info[key] = value
+
+    def update_signature_and_abbr(self, key:str, abbr:str, args:dict):
+        self._abbreviated.update({
+            key: abbr
+        })
+
+        self.signature_info.update({
+            key: args.get(key, None)
+        })
+
+    def format(self, short: bool = False) -> str:
+        pairs = []
+        keys = list(self.signature_info.keys())
+        for name in keys:
+            value = self.signature_info[name]
+            if value is not None:
+                # Check for nested signature objects
+                if hasattr(value, "get_signature"):
+
+                    # Wrap nested signatures in brackets
+                    nested_signature = value.get_signature()
+                    if isinstance(nested_signature, Signature):
+                        nested_signature = nested_signature.format(short=short)
+                    value = f"{{{nested_signature}}}"
+                if isinstance(value, bool):
+                    # Replace True/False with yes/no
+                    value = "yes" if value else "no"
+                if isinstance(value, Callable):
+                    value = value.__name__
+                final_name = self._abbreviated[name] if short else name
+                pairs.append(f"{final_name}:{value}")
+
+        return "|".join(pairs)
+
+    def __str__(self):
+        return self.format()
+
+    def __repr__(self):
+        return self.format()
+
+
+class SignatureMixin:
+    _SIGNATURE_TYPE = Signature
+    def get_signature(self) -> Signature:
+        return self._SIGNATURE_TYPE(self.__dict__)
 
 class BaseMetric[T]:
     """Base class for all metrics."""
+    # Each metric should define its Signature class' name here
+    _SIGNATURE_TYPE = Signature
 
-    def __init__(self, name: str, higher_is_better: bool = True):
+    def __init__(self, name: str, higher_is_better: bool = False):
         self.name = name
         self.higher_is_better = higher_is_better
 
@@ -38,3 +103,8 @@ def score_all(self, hypotheses: list[T], references: list[T], progress_bar=True)
 
     def __str__(self):
         return self.name
+
+    def get_signature(self) -> Signature:
+        return self._SIGNATURE_TYPE(self.__dict__)
+
+

pose_evaluation/metrics/base_pose_metric.py

@@ -1,5 +1,55 @@
+from typing import Iterable, List, cast, Union
 from pose_format import Pose
+from pose_evaluation.metrics.pose_processors import PoseProcessor
 
-from pose_evaluation.metrics.base import BaseMetric
+from pose_evaluation.metrics.base import BaseMetric, Signature
 
-PoseMetric = BaseMetric[Pose]
+
+class PoseMetricSignature(Signature):
+
+    def __init__(self, args: dict):
+        super().__init__(args)
+
+        self._abbreviated.update({"pose_preprocessers": "pre"})
+
+        pose_preprocessors = args.get("pose_preprocessers", None)
+        prep_string = ""
+        if pose_preprocessors is not None:
+            prep_string = (
+                "{" + "|".join([f"{prep}" for prep in pose_preprocessors]) + "}"
+            )
+
+        self.signature_info.update(
+            {"pose_preprocessers": prep_string if pose_preprocessors else None}
+        )
+
+
+class PoseMetric(BaseMetric[Pose]):
+    _SIGNATURE_TYPE = PoseMetricSignature
+
+    def __init__(
+        self,
+        name: str = "PoseMetric",
+        higher_is_better: bool = False,
+        pose_preprocessors: Union[None, List[PoseProcessor]] = None,
+    ):
+
+        super().__init__(name, higher_is_better)
+        if pose_preprocessors is None:
+            self.pose_preprocessers = []
+        else:
+            self.pose_preprocessers = pose_preprocessors
+
+    def score(self, hypothesis: Pose, reference: Pose) -> float:
+        hypothesis, reference = self.process_poses([hypothesis, reference])
+        return self.score(hypothesis, reference)
+
+    def process_poses(self, poses: Iterable[Pose]) -> List[Pose]:
+        poses = list(poses)
+        for preprocessor in self.pose_preprocessers:
+            preprocessor = cast(PoseProcessor, preprocessor)
+            poses = preprocessor.process_poses(poses)
+        return poses
+
+    def add_preprocessor(self, processor: PoseProcessor):
+        self.pose_preprocessers.append(processor)

pose_evaluation/metrics/build_ham2pose_metrics.py

@@ -0,0 +1,94 @@
+from pose_evaluation.metrics.distance_measure import (
+    PowerDistance,
+)
+from pose_evaluation.metrics.distance_metric import DistanceMetric
+from pose_evaluation.metrics.ham2pose_distances import (
+    Ham2PoseMSEDistance,
+    Ham2PoseMaskedEuclideanDistance,
+    Ham2PoseAPEDistance,
+)
+from pose_evaluation.metrics.mje_metric import MeanJointErrorMetric
+from pose_evaluation.metrics.dynamic_time_warping_metric import DTWMetric
+from pose_evaluation.metrics.pose_processors import (
+    get_standard_pose_processors,
+)
+
+if __name__ == "__main__":
+
+    metrics = []
+    MJEMetric = (
+        MeanJointErrorMetric()
+    )  # automatically sets distance measure, zero-padding.
+    metrics.append(MJEMetric)
+
+    Ham2Pose_DTW_MJE_Metric = DTWMetric(
+        name="DTW_MJE",
+        distance_measure=PowerDistance(2, 0),
+        pose_preprocessors=get_standard_pose_processors(zero_pad_shorter=False),
+    )
+
+    Ham2Pose_nDTW_MJE_Metric = DTWMetric(
+        name="nDTW_MJE",
+        distance_measure=Ham2PoseMaskedEuclideanDistance(),
+        pose_preprocessors=get_standard_pose_processors(zero_pad_shorter=False),
+    )
+
+    metrics.append(Ham2Pose_DTW_MJE_Metric)
+    metrics.append(Ham2Pose_nDTW_MJE_Metric)
+
+    # Ham2Pose APE is a PowerDistance. But with a few preprocessors.
+    # 1. standard preprocessors
+    # 2. then these: basically this is "zero_pad_shorter", and also setting masked values to zero.
+    # if len(trajectory1) < len(trajectory2):
+    #     diff = len(trajectory2) - len(trajectory1)
+    #     trajectory1 = np.concatenate((trajectory1, np.zeros((diff, 3))))
+    # elif len(trajectory2) < len(trajectory1):
+    #     trajectory2 = np.concatenate((trajectory2, np.zeros((len(trajectory1) - len(trajectory2), 3))))
+    # pose1_mask = np.ma.getmask(trajectory1)
+    # pose2_mask = np.ma.getmask(trajectory2)
+    # trajectory1[pose1_mask] = 0
+    # trajectory1[pose2_mask] = 0
+    # trajectory2[pose1_mask] = 0
+    # trajectory2[pose2_mask] = 0
+    # 3. pointwise aggregate by SUM
+    # sq_error = np.power(trajectory1 - trajectory2, 2).sum(-1)
+    # 4. trajectorywise aggregate by MEAN
+    # np.sqrt(sq_error).mean()
+    Ham2PoseAPEMetric = DistanceMetric(
+        name="Ham2PoseAPEMetric",
+        distance_measure=Ham2PoseAPEDistance(),
+        pose_preprocessors=get_standard_pose_processors(
+            zero_pad_shorter=True, set_masked_values_to_origin=True
+        ),
+    )
+    metrics.append(Ham2PoseAPEMetric)
+
+    # MSE from Ham2Pose is zero-padding, plus set to origin, and then squared error.
+    # if len(trajectory1) < len(trajectory2):
+    #     diff = len(trajectory2) - len(trajectory1)
+    #     trajectory1 = np.concatenate((trajectory1, np.zeros((diff, 3))))
+    # elif len(trajectory2) < len(trajectory1):
+    #     trajectory2 = np.concatenate((trajectory2, np.zeros((len(trajectory1) - len(trajectory2), 3))))
+    # pose1_mask = np.ma.getmask(trajectory1)
+    # pose2_mask = np.ma.getmask(trajectory2)
+    # trajectory1[pose1_mask] = 0
+    # trajectory1[pose2_mask] = 0
+    # trajectory2[pose1_mask] = 0
+    # trajectory2[pose2_mask] = 0
+    # sq_error = np.power(trajectory1 - trajectory2, 2).sum(-1)
+    # return sq_error.mean()
+    Ham2PoseMSEMetric = DistanceMetric(
+        name="Ham2PoseMSEMetric",
+        distance_measure=Ham2PoseMSEDistance(),
+        pose_preprocessors=get_standard_pose_processors(
+            zero_pad_shorter=True, set_masked_values_to_origin=True
+        ),
+    )
+    metrics.append(Ham2PoseMSEMetric)
+
+    for metric in metrics:
+        print("*" * 30)
+        print(f"METRIC: {metric}")
+        print(metric.get_signature())
+        print(metric.get_signature().format(short=True))
+        print()