mmpose.codecs.associative_embedding 源代码

# Copyright (c) OpenMMLab. All rights reserved.
from collections import namedtuple
from itertools import product
from typing import Any, List, Optional, Tuple

import numpy as np
import torch
from munkres import Munkres
from torch import Tensor

from mmpose.registry import KEYPOINT_CODECS
from mmpose.utils.tensor_utils import to_numpy
from .base import BaseKeypointCodec
from .utils import (batch_heatmap_nms, generate_gaussian_heatmaps,
                    generate_udp_gaussian_heatmaps, refine_keypoints,

def _group_keypoints_by_tags(vals: np.ndarray,
                             tags: np.ndarray,
                             locs: np.ndarray,
                             keypoint_order: List[int],
                             val_thr: float,
                             tag_thr: float = 1.0,
                             max_groups: Optional[int] = None) -> np.ndarray:
    """Group the keypoints by tags using Munkres algorithm.


        - keypoint number: K
        - candidate number: M
        - tag dimenssion: L
        - coordinate dimension: D
        - group number: G

        vals (np.ndarray): The heatmap response values of keypoints in shape
            (K, M)
        tags (np.ndarray): The tags of the keypoint candidates in shape
            (K, M, L)
        locs (np.ndarray): The locations of the keypoint candidates in shape
            (K, M, D)
        keypoint_order (List[int]): The grouping order of the keypoints.
            The groupping usually starts from a keypoints around the head and
            torso, and gruadually moves out to the limbs
        val_thr (float): The threshold of the keypoint response value
        tag_thr (float): The maximum allowed tag distance when matching a
            keypoint to a group. A keypoint with larger tag distance to any
            of the existing groups will initializes a new group
        max_groups (int, optional): The maximum group number. ``None`` means
            no limitation. Defaults to ``None``

        np.ndarray: grouped keypoints in shape (G, K, D+1), where the last
        dimenssion is the concatenated keypoint coordinates and scores.
    K, M, D = locs.shape
    assert vals.shape == tags.shape[:2] == (K, M)
    assert len(keypoint_order) == K

    # Build Munkres instance
    munkres = Munkres()

    # Build a group pool, each group contains the keypoints of an instance
    groups = []

    Group = namedtuple('Group', field_names=['kpts', 'scores', 'tag_list'])

    def _init_group():
        """Initialize a group, which is composed of the keypoints, keypoint
        scores and the tag of each keypoint."""
        _group = Group(
            kpts=np.zeros((K, D), dtype=np.float32),
            scores=np.zeros(K, dtype=np.float32),
        return _group

    for i in keypoint_order:
        # Get all valid candidate of the i-th keypoints
        valid = vals[i] > val_thr
        if not valid.any():

        tags_i = tags[i, valid]  # (M', L)
        vals_i = vals[i, valid]  # (M',)
        locs_i = locs[i, valid]  # (M', D)

        if len(groups) == 0:  # Initialize the group pool
            for tag, val, loc in zip(tags_i, vals_i, locs_i):
                group = _init_group()
                group.kpts[i] = loc
                group.scores[i] = val


        else:  # Match keypoints to existing groups
            groups = groups[:max_groups]
            group_tags = [np.mean(g.tag_list, axis=0) for g in groups]

            # Calculate distance matrix between group tags and tag candidates
            # of the i-th keypoint
            # Shape: (M', 1, L) , (1, G, L) -> (M', G, L)
            diff = tags_i[:, None] - np.array(group_tags)[None]
            dists = np.linalg.norm(diff, ord=2, axis=2)
            num_kpts, num_groups = dists.shape[:2]

            # Experimental cost function for keypoint-group matching
            costs = np.round(dists) * 100 - vals_i[..., None]
            if num_kpts > num_groups:
                padding = np.full((num_kpts, num_kpts - num_groups),
                costs = np.concatenate((costs, padding), axis=1)

            # Match keypoints and groups by Munkres algorithm
            matches = munkres.compute(costs)
            for kpt_idx, group_idx in matches:
                if group_idx < num_groups and dists[kpt_idx,
                                                    group_idx] < tag_thr:
                    # Add the keypoint to the matched group
                    group = groups[group_idx]
                    # Initialize a new group with unmatched keypoint
                    group = _init_group()

                group.kpts[i] = locs_i[kpt_idx]
                group.scores[i] = vals_i[kpt_idx]

    groups = groups[:max_groups]
    if groups:
        grouped_keypoints = np.stack(
            [np.r_['1', g.kpts, g.scores[:, None]] for g in groups])
        grouped_keypoints = np.empty((0, K, D + 1))

    return grouped_keypoints

[文档]@KEYPOINT_CODECS.register_module() class AssociativeEmbedding(BaseKeypointCodec): """Encode/decode keypoints with the method introduced in "Associative Embedding". This is an asymmetric codec, where the keypoints are represented as gaussian heatmaps and position indices during encoding, and restored from predicted heatmaps and group tags. See the paper `Associative Embedding: End-to-End Learning for Joint Detection and Grouping`_ by Newell et al (2017) for details Note: - instance number: N - keypoint number: K - keypoint dimension: D - embedding tag dimension: L - image size: [w, h] - heatmap size: [W, H] Encoded: - heatmaps (np.ndarray): The generated heatmap in shape (K, H, W) where [W, H] is the `heatmap_size` - keypoint_indices (np.ndarray): The keypoint position indices in shape (N, K, 2). Each keypoint's index is [i, v], where i is the position index in the heatmap (:math:`i=y*w+x`) and v is the visibility - keypoint_weights (np.ndarray): The target weights in shape (N, K) Args: input_size (tuple): Image size in [w, h] heatmap_size (tuple): Heatmap size in [W, H] sigma (float): The sigma value of the Gaussian heatmap use_udp (bool): Whether use unbiased data processing. See `UDP (CVPR 2020)`_ for details. Defaults to ``False`` decode_keypoint_order (List[int]): The grouping order of the keypoint indices. The groupping usually starts from a keypoints around the head and torso, and gruadually moves out to the limbs decode_keypoint_thr (float): The threshold of keypoint response value in heatmaps. Defaults to 0.1 decode_tag_thr (float): The maximum allowed tag distance when matching a keypoint to a group. A keypoint with larger tag distance to any of the existing groups will initializes a new group. Defaults to 1.0 decode_nms_kernel (int): The kernel size of the NMS during decoding, which should be an odd integer. Defaults to 5 decode_gaussian_kernel (int): The kernel size of the Gaussian blur during decoding, which should be an odd integer. It is only used when ``self.use_udp==True``. Defaults to 3 decode_topk (int): The number top-k candidates of each keypoints that will be retrieved from the heatmaps during dedocding. Defaults to 20 decode_max_instances (int, optional): The maximum number of instances to decode. ``None`` means no limitation to the instance number. Defaults to ``None`` .. _`Associative Embedding: End-to-End Learning for Joint Detection and Grouping`: .. _`UDP (CVPR 2020)`: """ def __init__( self, input_size: Tuple[int, int], heatmap_size: Tuple[int, int], sigma: Optional[float] = None, use_udp: bool = False, decode_keypoint_order: List[int] = [], decode_nms_kernel: int = 5, decode_gaussian_kernel: int = 3, decode_keypoint_thr: float = 0.1, decode_tag_thr: float = 1.0, decode_topk: int = 20, decode_max_instances: Optional[int] = None, ) -> None: super().__init__() self.input_size = input_size self.heatmap_size = heatmap_size self.use_udp = use_udp self.decode_nms_kernel = decode_nms_kernel self.decode_gaussian_kernel = decode_gaussian_kernel self.decode_keypoint_thr = decode_keypoint_thr self.decode_tag_thr = decode_tag_thr self.decode_topk = decode_topk self.decode_max_instances = decode_max_instances self.dedecode_keypoint_order = decode_keypoint_order.copy() if self.use_udp: self.scale_factor = ((np.array(input_size) - 1) / (np.array(heatmap_size) - 1)).astype( np.float32) else: self.scale_factor = (np.array(input_size) / heatmap_size).astype(np.float32) if sigma is None: sigma = (heatmap_size[0] * heatmap_size[1])**0.5 / 64 self.sigma = sigma
[文档] def encode( self, keypoints: np.ndarray, keypoints_visible: Optional[np.ndarray] = None ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """Encode keypoints into heatmaps and position indices. Note that the original keypoint coordinates should be in the input image space. Args: keypoints (np.ndarray): Keypoint coordinates in shape (N, K, D) keypoints_visible (np.ndarray): Keypoint visibilities in shape (N, K) Returns: dict: - heatmaps (np.ndarray): The generated heatmap in shape (K, H, W) where [W, H] is the `heatmap_size` - keypoint_indices (np.ndarray): The keypoint position indices in shape (N, K, 2). Each keypoint's index is [i, v], where i is the position index in the heatmap (:math:`i=y*w+x`) and v is the visibility - keypoint_weights (np.ndarray): The target weights in shape (N, K) """ if keypoints_visible is None: keypoints_visible = np.ones(keypoints.shape[:2], dtype=np.float32) # keypoint coordinates in heatmap _keypoints = keypoints / self.scale_factor if self.use_udp: heatmaps, keypoint_weights = generate_udp_gaussian_heatmaps( heatmap_size=self.heatmap_size, keypoints=_keypoints, keypoints_visible=keypoints_visible, sigma=self.sigma) else: heatmaps, keypoint_weights = generate_gaussian_heatmaps( heatmap_size=self.heatmap_size, keypoints=_keypoints, keypoints_visible=keypoints_visible, sigma=self.sigma) keypoint_indices = self._encode_keypoint_indices( heatmap_size=self.heatmap_size, keypoints=_keypoints, keypoints_visible=keypoints_visible) encoded = dict( heatmaps=heatmaps, keypoint_indices=keypoint_indices, keypoint_weights=keypoint_weights) return encoded
def _encode_keypoint_indices(self, heatmap_size: Tuple[int, int], keypoints: np.ndarray, keypoints_visible: np.ndarray) -> np.ndarray: w, h = heatmap_size N, K, _ = keypoints.shape keypoint_indices = np.zeros((N, K, 2), dtype=np.int64) for n, k in product(range(N), range(K)): x, y = (keypoints[n, k] + 0.5).astype(np.int64) index = y * w + x vis = (keypoints_visible[n, k] > 0.5 and 0 <= x < w and 0 <= y < h) keypoint_indices[n, k] = [index, vis] return keypoint_indices
[文档] def decode(self, encoded: Any) -> Tuple[np.ndarray, np.ndarray]: raise NotImplementedError()
def _get_batch_topk(self, batch_heatmaps: Tensor, batch_tags: Tensor, k: int): """Get top-k response values from the heatmaps and corresponding tag values from the tagging heatmaps. Args: batch_heatmaps (Tensor): Keypoint detection heatmaps in shape (B, K, H, W) batch_tags (Tensor): Tagging heatmaps in shape (B, C, H, W), where the tag dim C is 2*K when using flip testing, or K otherwise k (int): The number of top responses to get Returns: tuple: - topk_vals (Tensor): Top-k response values of each heatmap in shape (B, K, Topk) - topk_tags (Tensor): The corresponding embedding tags of the top-k responses, in shape (B, K, Topk, L) - topk_locs (Tensor): The location of the top-k responses in each heatmap, in shape (B, K, Topk, 2) where last dimension represents x and y coordinates """ B, K, H, W = batch_heatmaps.shape L = batch_tags.shape[1] // K # shape of topk_val, top_indices: (B, K, TopK) topk_vals, topk_indices = batch_heatmaps.flatten(-2, -1).topk( k, dim=-1) topk_tags_per_kpts = [ torch.gather(_tag, dim=2, index=topk_indices) for _tag in torch.unbind(batch_tags.view(B, L, K, H * W), dim=1) ] topk_tags = torch.stack(topk_tags_per_kpts, dim=-1) # (B, K, TopK, L) topk_locs = torch.stack([topk_indices % W, topk_indices // W], dim=-1) # (B, K, TopK, 2) return topk_vals, topk_tags, topk_locs def _group_keypoints(self, batch_vals: np.ndarray, batch_tags: np.ndarray, batch_locs: np.ndarray): """Group keypoints into groups (each represents an instance) by tags. Args: batch_vals (Tensor): Heatmap response values of keypoint candidates in shape (B, K, Topk) batch_tags (Tensor): Tags of keypoint candidates in shape (B, K, Topk, L) batch_locs (Tensor): Locations of keypoint candidates in shape (B, K, Topk, 2) Returns: List[np.ndarray]: Grouping results of a batch, each element is a np.ndarray (in shape [N, K, D+1]) that contains the groups detected in an image, including both keypoint coordinates and scores. """ def _group_func(inputs: Tuple): vals, tags, locs = inputs return _group_keypoints_by_tags( vals, tags, locs, keypoint_order=self.dedecode_keypoint_order, val_thr=self.decode_keypoint_thr, tag_thr=self.decode_tag_thr, max_groups=self.decode_max_instances) _results = map(_group_func, zip(batch_vals, batch_tags, batch_locs)) results = list(_results) return results def _fill_missing_keypoints(self, keypoints: np.ndarray, keypoint_scores: np.ndarray, heatmaps: np.ndarray, tags: np.ndarray): """Fill the missing keypoints in the initial predictions. Args: keypoints (np.ndarray): Keypoint predictions in shape (N, K, D) keypoint_scores (np.ndarray): Keypint score predictions in shape (N, K), in which 0 means the corresponding keypoint is missing in the initial prediction heatmaps (np.ndarry): Heatmaps in shape (K, H, W) tags (np.ndarray): Tagging heatmaps in shape (C, H, W) where C=L*K Returns: tuple: - keypoints (np.ndarray): Keypoint predictions with missing ones filled - keypoint_scores (np.ndarray): Keypoint score predictions with missing ones filled """ N, K = keypoints.shape[:2] H, W = heatmaps.shape[1:] L = tags.shape[0] // K keypoint_tags = [tags[k::K] for k in range(K)] for n in range(N): # Calculate the instance tag (mean tag of detected keypoints) _tag = [] for k in range(K): if keypoint_scores[n, k] > 0: x, y = keypoints[n, k, :2].astype(np.int64) x = np.clip(x, 0, W - 1) y = np.clip(y, 0, H - 1) _tag.append(keypoint_tags[k][:, y, x]) tag = np.mean(_tag, axis=0) tag = tag.reshape(L, 1, 1) # Search maximum response of the missing keypoints for k in range(K): if keypoint_scores[n, k] > 0: continue dist_map = np.linalg.norm( keypoint_tags[k] - tag, ord=2, axis=0) cost_map = np.round(dist_map) * 100 - heatmaps[k] # H, W y, x = np.unravel_index(np.argmin(cost_map), shape=(H, W)) keypoints[n, k] = [x, y] keypoint_scores[n, k] = heatmaps[k, y, x] return keypoints, keypoint_scores
[文档] def batch_decode(self, batch_heatmaps: Tensor, batch_tags: Tensor ) -> Tuple[List[np.ndarray], List[np.ndarray]]: """Decode the keypoint coordinates from a batch of heatmaps and tagging heatmaps. The decoded keypoint coordinates are in the input image space. Args: batch_heatmaps (Tensor): Keypoint detection heatmaps in shape (B, K, H, W) batch_tags (Tensor): Tagging heatmaps in shape (B, C, H, W), where :math:`C=L*K` Returns: tuple: - batch_keypoints (List[np.ndarray]): Decoded keypoint coordinates of the batch, each is in shape (N, K, D) - batch_scores (List[np.ndarray]): Decoded keypoint scores of the batch, each is in shape (N, K). It usually represents the confidience of the keypoint prediction """ B, _, H, W = batch_heatmaps.shape assert batch_tags.shape[0] == B and batch_tags.shape[2:4] == (H, W), ( f'Mismatched shapes of heatmap ({batch_heatmaps.shape}) and ' f'tagging map ({batch_tags.shape})') # Heatmap NMS batch_heatmaps = batch_heatmap_nms(batch_heatmaps, self.decode_nms_kernel) # Get top-k in each heatmap and and convert to numpy batch_topk_vals, batch_topk_tags, batch_topk_locs = to_numpy( self._get_batch_topk( batch_heatmaps, batch_tags, k=self.decode_topk)) # Group keypoint candidates into groups (instances) batch_groups = self._group_keypoints(batch_topk_vals, batch_topk_tags, batch_topk_locs) # Convert to numpy batch_heatmaps_np = to_numpy(batch_heatmaps) batch_tags_np = to_numpy(batch_tags) # Refine the keypoint prediction batch_keypoints = [] batch_keypoint_scores = [] for i, (groups, heatmaps, tags) in enumerate( zip(batch_groups, batch_heatmaps_np, batch_tags_np)): keypoints, scores = groups[..., :-1], groups[..., -1] if keypoints.size > 0: # identify missing keypoints keypoints, scores = self._fill_missing_keypoints( keypoints, scores, heatmaps, tags) # refine keypoint coordinates according to heatmap distribution if self.use_udp: keypoints = refine_keypoints_dark_udp( keypoints, heatmaps, blur_kernel_size=self.decode_gaussian_kernel) else: keypoints = refine_keypoints(keypoints, heatmaps) batch_keypoints.append(keypoints) batch_keypoint_scores.append(scores) # restore keypoint scale batch_keypoints = [ kpts * self.scale_factor for kpts in batch_keypoints ] return batch_keypoints, batch_keypoint_scores
Read the Docs v: latest
On Read the Docs
Project Home

Free document hosting provided by Read the Docs.