(3e -> 3a) Add InstanceGroup class

sleap/gui/commands.py

@@ -35,6 +35,7 @@ class which inherits from `AppCommand` (or a more specialized class such as
 import traceback
 from enum import Enum
 from glob import glob
+from itertools import permutations, product
 from pathlib import Path, PurePath
 from typing import Callable, Dict, Iterator, List, Optional, Tuple, Type, Union, cast
 
@@ -3750,6 +3751,210 @@ def get_all_views_at_frame(
 
         return views
 
+    @staticmethod
+    def get_instance_grouping(
+        instances: Dict[int, Dict[Camcorder, List[Instance]]],
+        reprojection_error_per_frame: Dict[int, float],
+    ) -> Dict[int, Dict[Camcorder, List[Instance]]]:
+        """Get instance grouping for triangulation."""
+
+        frame_with_min_error = min(
+            reprojection_error_per_frame, key=reprojection_error_per_frame.get
+        )
+
+        best_instances = instances[frame_with_min_error]
+        best_instances_correct_format = {frame_with_min_error: best_instances}
+
+        return best_instances_correct_format
+
+    @staticmethod
+    def _calculate_reprojection_error(
+        session: RecordingSession,
+        instances: Dict[int, Dict[Camcorder, List[Instance]]],
+        per_instance: bool = False,
+        per_view: bool = False,
+    ) -> Union[
+        Dict[int, float], Dict[int, Dict[Camcorder, List[Tuple[Instance, float]]]]
+    ]:
+        """Calculate reprojection error per frame or per instance.
+
+        Args:
+            session: The `RecordingSession` containing the `Camcorder`s.
+            instances: Dict with frame identifier keys (not the frame index) and values
+                of another inner dict with `Camcorder` keys and `List[Instance]` values.
+            per_instance: If True, then return a dict with frame identifier keys and
+                values of another inner dict with `Camcorder` keys and
+                `List[Tuple[Instance, float]]` values.
+            per_view: If True, then return a dict with frame identifier keys and values
+                of another inner dict with `Camcorder` keys and
+                `Tuple[Tuple[str, str], float]` values. If per_instance is True, then that takes precendence.
+
+        Returns:
+            Dict with frame identifier keys (not the frame index) and values of another
+            inner dict with `Camcorder` keys and `List[Tuple[Instance, float]]` values
+            if per_instance is True, otherwise a dict with frame identifier keys and
+            values of reprojection error for the frame.
+        """
+
+        reprojection_error_per_frame = {}
+
+        # Triangulate and reproject instance coordinates.
+        instances_and_coords: Dict[
+            int, Dict[Camcorder, Iterator[Tuple[Instance, np.ndarray]]]
+        ] = TriangulateSession.calculate_reprojected_points(
+            session=session, instances=instances
+        )
+        for frame_id, instances_in_frame in instances_and_coords.items():
+            frame_error = {} if per_instance or per_view else 0
+            for cam, instances_in_view in instances_in_frame.items():
+                # Compare instance coordinates here
+                instance_ids = []
+                view_error = [] if per_instance else 0
+                for inst, inst_coords in instances_in_view:
+                    node_errors = np.nan_to_num(inst.numpy() - inst_coords)
+                    instance_error = np.linalg.norm(node_errors)
+
+                    if per_instance:
+                        view_error.append((inst, instance_error))
+                    else:
+                        view_error += instance_error
+
+                    inst_id = inst.track if inst.track is not None else "None"
+                    instance_ids.append(inst_id)
+
+                if per_instance:
+                    frame_error[cam] = view_error
+                elif per_view:
+                    frame_error[cam] = (tuple(instance_ids), view_error)
+                else:
+                    frame_error += view_error
+
+            reprojection_error_per_frame[frame_id] = frame_error
+
+        return reprojection_error_per_frame
+
+    @staticmethod
+    def calculate_error_per_instance(
+        session: RecordingSession, instances: Dict[int, Dict[Camcorder, List[Instance]]]
+    ) -> Dict[int, float]:
+        """Calculate reprojection error per instance."""
+
+        reprojection_error_per_instance = (
+            TriangulateSession._calculate_reprojection_error(
+                session=session, instances=instances, per_instance=True
+            )
+        )
+
+        return reprojection_error_per_instance
+
+    @staticmethod
+    def calculate_error_per_view(
+        session: RecordingSession, instances: Dict[int, Dict[Camcorder, List[Instance]]]
+    ) -> Dict[int, float]:
+        """Calculate reprojection error per instance."""
+
+        reprojection_error_per_view = TriangulateSession._calculate_reprojection_error(
+            session=session, instances=instances, per_view=True
+        )
+
+        return reprojection_error_per_view
+
+    @staticmethod
+    def calculate_error_per_frame(
+        session: RecordingSession, instances: Dict[int, Dict[Camcorder, List[Instance]]]
+    ) -> Dict[int, float]:
+        """Calculate reprojection error per frame."""
+
+        reprojection_error_per_frame = TriangulateSession._calculate_reprojection_error(
+            session=session, instances=instances, per_instance=False
+        )
+
+        return reprojection_error_per_frame
+
+    @staticmethod
+    def get_products_of_instances(
+        selected_instance: Instance,
+        session: RecordingSession,
+        frame_idx: int,
+        cams_to_include: Optional[List[Camcorder]] = None,
+    ) -> Dict[int, Dict[Camcorder, List[Instance]]]:
+        """Get all (single-instance) possible products of instances across views.
+
+        Args:
+            selected_instance: The `Instance` to add to  permutations of instances in
+                views other than that of the `selected_instance`.
+            session: The `RecordingSession` containing the `Camcorder`s.
+            frame_idx: Frame index to get instances from (0-indexed).
+            cams_to_include: List of `Camcorder`s to include. Default is all.
+            require_multiple_views: If True, then raise and error if one or less views
+                or instances are found.
+
+        Raises:
+            ValueError if one or less views or instances are found.
+
+        Returns:
+            Dict with frame identifier keys (not the frame index) and values of another
+            inner dict with `Camcorder` keys and `List[Instance]` values. Each
+            `List[Instance]` is of length 1.
+        """
+
+        cam_selected = session.get_camera(selected_instance.video)
+        cam_selected = cast(Camcorder, cam_selected)  # Could be None if not in session
+
+        # Get all instances accross views at this frame index, then remove selected
+        instances: Dict[
+            Camcorder, List[Instance]
+        ] = TriangulateSession.get_instances_across_views(
+            session=session,
+            frame_idx=frame_idx,
+            cams_to_include=cams_to_include,
+            track=-1,  # Get all instances regardless of track.
+            require_multiple_views=True,
+        )
+
+        # Find max number of instances in other views
+        max_num_instances = max([len(instances) for instances in instances.values()])
+
+        # Create a dummy instance of all nan values
+        dummy_instance = Instance.from_numpy(
+            np.full(
+                shape=(len(selected_instance.skeleton.nodes), 2),
+                fill_value=np.nan,
+            ),
+            skeleton=selected_instance.skeleton,
+        )
+
+        # Get permutations of instances from other views
+        instances_permutations: Dict[Camcorder, Iterator[Tuple]] = {}
+        for cam, instances_in_view in instances.items():
+            # Append a dummy instance to all lists of instances if less than the max length
+            num_missing = 1
+            num_instances = len(instances_in_view)
+            if num_instances < max_num_instances:
+                num_missing = max_num_instances - num_instances
+
+                # Extend the list first
+                instances_in_view.extend([dummy_instance] * num_missing)
+
+            # Permute instances into all possible orderings w/in a view
+            instances_permutations[cam] = permutations(instances_in_view)
+
+        # Get products of instances from other views into all possible groupings
+        # Ordering of dict_values is preserved in Python 3.7+
+        products_of_instances: Iterator[Iterator[Tuple]] = product(
+            *instances_permutations.values()
+        )
+
+        # Reorganize products by cam and add selected instance to each permutation
+        instances_hypotheses: Dict[int, Dict[Camcorder, List[Instance]]] = {}
+        for frame_id, prod in enumerate(products_of_instances):
+            instances_hypotheses[frame_id] = {
+                cam: [*inst] for cam, inst in zip(instances.keys(), prod)
+            }
+
+        # Expect "max # instances in view" ** "# views" frames (a.k.a. hypotheses)
+        return instances_hypotheses
+
     @staticmethod
     def get_instances_matrices(
         instances: Dict[int, Dict[Camcorder, List[Instance]]],
@@ -3804,7 +4009,9 @@ def get_instances_matrices(
             inst_coords_frames.append(
                 inst_coords_views
             )  # len=frame_idx, List[M x T x N x 2]
+
         inst_coords = np.stack(inst_coords_frames, axis=1)  # M x F x T x N x 2
+        cams_ordered = cast(List[Camcorder], cams_ordered)  # Could be None if no frames
 
         return inst_coords, cams_ordered
 

sleap/io/cameras.py

@@ -2,15 +2,14 @@
 import logging
 import tempfile
 from pathlib import Path
-from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
+from typing import Any, Dict, Iterator, List, Optional, Tuple, Union, cast
 
 import cattr
 import numpy as np
 import toml
 from aniposelib.cameras import Camera, CameraGroup, FisheyeCamera
 from attrs import define, field
 from attrs.validators import deep_iterable, instance_of
-from sleap_anipose import reproject, triangulate
 
 # from sleap.io.dataset import Labels  # TODO(LM): Circular import, implement Observer
 from sleap.io.video import Video
@@ -755,3 +754,146 @@ def make_cattr(videos_list: List[Video]):
             RecordingSession, lambda x: x.to_session_dict(video_to_idx)
         )
         return sessions_cattr
+
+
+@define
+class InstanceGroup:
+    """Defines a group of instances across the same frame index.
+
+    Args:
+        camera_cluster: `CameraCluster` object.
+        instances: List of `Instance` objects.
+
+    """
+
+    frame_idx: int = field(validator=instance_of(int))
+    camera_cluster: Optional[CameraCluster] = None
+    _instance_by_camcorder: Dict[Camcorder, "Instance"] = field(factory=dict)
+    _camcorder_by_instance: Dict["Instance", Camcorder] = field(factory=dict)
+
+    def __attrs_post_init__(self):
+        """Initialize `InstanceGroup` object."""
+
+        for cam, instance in self._instance_by_camcorder.items():
+            self._camcorder_by_instance[instance] = cam
+
+    @property
+    def instances(self) -> List["Instance"]:
+        """List of `Instance` objects."""
+        return list(self._instance_by_camcorder.values())
+
+    def get_instance(self, cam: Camcorder) -> Optional["Instance"]:
+        """Retrieve `Instance` linked to `Camcorder`.
+
+        Args:
+            camcorder: `Camcorder` object.
+
+        Returns:
+            If `Camcorder` in `self.camera_cluster`, then `Instance` object if found, else
+            `None` if `Camcorder` has no linked `Instance`.
+        """
+
+        if cam not in self._instance_by_camcorder:
+            logger.warning(
+                f"Camcorder {cam.name} is not linked to a video in this "
+                f"RecordingSession."
+            )
+            return None
+
+        return self._instance_by_camcorder[cam]
+
+    def get_cam(self, instance: "Instance") -> Optional[Camcorder]:
+        """Retrieve `Camcorder` linked to `Instance`.
+
+        Args:
+            instance: `Instance` object.
+
+        Returns:
+            `Camcorder` object if found, else `None`.
+        """
+
+        if instance not in self._camcorder_by_instance:
+            logger.warning(
+                f"{instance} is not in this InstanceGroup's Instances: \n\t{self.instances}."
+            )
+            return None
+
+        return self._camcorder_by_instance[instance]
+
+    def __getitem__(
+        self, idx_or_key: Union[int, Camcorder, "Instance"]
+    ) -> Union[Camcorder, "Instance"]:
+        """Grab a `Camcorder` of `Instance` from the `InstanceGroup`."""
+
+        # Try to find in `self.camera_cluster.cameras`
+        if isinstance(idx_or_key, int):
+            try:
+                return self.instances[idx_or_key]
+            except IndexError:
+                pass
+
+        # Return a `Instance` if `idx_or_key` is a `Camcorder``
+        if isinstance(idx_or_key, Camcorder):
+            return self.get_instance(idx_or_key)
+
+        else:
+            # isinstance(idx_or_key, "Instance"):
+            try:
+                return self.get_cam(idx_or_key)
+            except:
+                pass
+
+        raise KeyError(
+            f"Key {idx_or_key} not found in {self.__class__.__name__} or "
+            "associated metadata."
+        )
+
+    def __len__(self):
+        return len(self.instances)
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}(frame_idx={self.frame_idx}, instances={len(self)}, camera_cluster={self.camera_cluster})"
+
+    @classmethod
+    def from_dict(cls, d: dict) -> "InstanceGroup":
+        """Creates an `InstanceGroup` object from a dictionary.
+
+        Args:
+            d: Dictionary with `Camcorder` keys and `Instance` values.
+
+        Returns:
+            `InstanceGroup` object.
+        """
+
+        frame_idx = None
+        for cam, instance in d.copy().items():
+            camera_cluster = cam.camera_cluster
+
+            # Remove dummy instances (determined by not having a frame index)
+            if instance.frame_idx is None:
+                d.pop(cam)
+            # Grab the frame index from non-dummy instances
+            elif frame_idx is None:
+                frame_idx = instance.frame_idx
+            # Ensure all instances have the same frame index
+            else:
+                try:
+                    assert frame_idx == instance.frame_idx
+                except AssertionError:
+                    logger.warning(
+                        f"Cannot create `InstanceGroup`: Frame index {frame_idx} "
+                        f"does not match instance frame index {instance.frame_idx}."
+                    )
+
+        if len(d) == 0:
+            logger.warning("Cannot create `InstanceGroup`: No real instances found.")
+
+        frame_idx = cast(
+            int, frame_idx
+        )  # Could be None if no real instances in dictionary
+
+        return cls(
+            frame_idx=frame_idx,
+            camera_cluster=camera_cluster,
+            instance_by_camcorder=d,
+        )