feat(tuner): add miner v1

finetuner/helper.py

@@ -14,6 +14,9 @@
 AnyDNN = TypeVar(
     'AnyDNN'
 )  #: The type of any implementation of a Deep Neural Network object
+AnyTensor = TypeVar(
+    'AnyTensor'
+)  #: The type of any implementation of an tensor for model tuning
 AnyDataLoader = TypeVar(
     'AnyDataLoader'
 )  #: The type of any implementation of a data loader

finetuner/tuner/base.py

@@ -1,14 +1,15 @@
 import abc
 import warnings
 from typing import (
+    Generator,
     Optional,
     Union,
     Tuple,
     List,
     Dict,
 )
 
-from ..helper import AnyDNN, AnyDataLoader, AnyOptimizer, DocumentArrayLike
+from ..helper import AnyDNN, AnyTensor, AnyDataLoader, AnyOptimizer, DocumentArrayLike
 from .summary import Summary
 
 
@@ -148,3 +149,15 @@ def __init__(
     ):
         super().__init__()
         self._inputs = inputs() if callable(inputs) else inputs
+
+
+class BaseMiner(abc.ABC):
+    @abc.abstractmethod
+    def mine(self, embeddings: List[AnyTensor], labels: List[int]):
+        """Generate tuples/triplets from input embeddings and labels, cut by limit if set.
+
+        :param embeddings: embeddings from model, should be a list of Tensor objects.
+        :param labels: labels of each embeddings, embeddings with same label indicates same class.
+        :return: tuple/triplet of embeddings.
+        """
+        ...

finetuner/tuner/pytorch/miner.py

@@ -0,0 +1,44 @@
+import torch
+import numpy as np
+from typing import List, Generator
+from itertools import combinations
+
+from ..base import BaseMiner
+
+
+class SiameseMiner(BaseMiner):
+    def mine(self, embeddings: List[torch.Tensor], labels: List[int]):
+        """Generate tuples from input embeddings and labels, cut by limit if set.
+
+        :param embeddings: embeddings from model, should be a list of Tensor objects.
+        :param labels: labels of each embeddings, embeddings with same label indicates same class.
+        :return: a pair of embeddings and their labels as tuple.
+        """
+        assert len(embeddings) == len(labels)
+        for left, right in combinations(enumerate(labels), 2):
+            if left[1] == right[1]:
+                yield embeddings[left[0]], embeddings[right[0]], 1
+            else:
+                yield embeddings[left[0]], embeddings[right[0]], -1
+
+
+class TripletMiner(BaseMiner):
+    def mine(self, embeddings: List[torch.Tensor], labels: List[int]):
+        """Generate triplets from input embeddings and labels, cut by limit if set.
+
+        :param embeddings: embeddings from model, should be a list of Tensor objects.
+        :param labels: labels of each embeddings, embeddings with same label indicates same class.
+        :return: triplet of embeddings follows the order of anchor, positive and negative.
+        """
+        assert len(embeddings) == len(labels)
+        labels1 = np.expand_dims(labels, 1)
+        labels2 = np.expand_dims(labels, 0)
+        matches = (labels1 == labels2).astype(int)
+        diffs = matches ^ 1
+        np.fill_diagonal(matches, 0)
+        triplets = np.expand_dims(matches, 2) * np.expand_dims(diffs, 1)
+        indices_left, indices_middle, indices_right = np.where(triplets)
+        for idx_left, idx_middle, idx_right in zip(
+            indices_left, indices_middle, indices_right
+        ):
+            yield embeddings[idx_left], embeddings[idx_middle], embeddings[idx_right]

tests/unit/tuner/torch/test_miner.py

@@ -0,0 +1,88 @@
+import pytest
+import torch
+
+from finetuner.tuner.pytorch.miner import SiameseMiner, TripletMiner
+
+BATCH_SIZE = 8
+NUM_DIM = 10
+
+
+@pytest.fixture
+def siamese_miner():
+    return SiameseMiner()
+
+
+@pytest.fixture
+def triplet_miner():
+    return TripletMiner()
+
+
+@pytest.fixture
+def embeddings():
+    return [torch.rand(NUM_DIM) for _ in range(BATCH_SIZE)]
+
+
+@pytest.fixture
+def labels():
+    return [1, 3, 1, 3, 2, 4, 2, 4]
+
+
+def _get_idx_by_tensor(embeddings, tensor):
+    for idx, embedding in enumerate(embeddings):
+        if torch.equal(tensor, embedding):
+            return idx
+
+
+def test_siamese_miner(embeddings, labels, siamese_miner):
+    rv = list(siamese_miner.mine(embeddings, labels))
+    assert len(rv) == 28
+    for item in rv:
+        tensor_left, tensor_right, label = item
+        tensor_left_idx = _get_idx_by_tensor(embeddings, tensor_left)
+        tensor_right_idx = _get_idx_by_tensor(embeddings, tensor_right)
+        # find corresponded label idx
+        tensor_left_label = labels[tensor_left_idx]
+        tensor_right_label = labels[tensor_right_idx]
+        if tensor_left_label == tensor_right_label:
+            expected_label = 1
+        else:
+            expected_label = -1
+        assert label == expected_label
+
+
+@pytest.mark.parametrize('cut_index', [0, 1])
+def test_siamese_miner_given_insufficient_inputs(
+    embeddings, labels, siamese_miner, cut_index
+):
+    embeddings = embeddings[:cut_index]
+    labels = labels[:cut_index]
+    rv = list(siamese_miner.mine(embeddings, labels))
+    assert len(rv) == 0
+
+
+def test_triplet_miner(embeddings, labels, triplet_miner):
+    rv = list(triplet_miner.mine(embeddings, labels))
+    assert len(rv) == 48
+    for item in rv:
+        tensor_left, tensor_middle, tensor_right = item
+        tensor_left_idx = _get_idx_by_tensor(embeddings, tensor_left)
+        tensor_middle_idx = _get_idx_by_tensor(embeddings, tensor_middle)
+        tensor_right_idx = _get_idx_by_tensor(embeddings, tensor_right)
+        # find corresponded label idx
+        tensor_left_label = labels[tensor_left_idx]
+        tensor_middle_label = labels[tensor_middle_idx]
+        tensor_right_label = labels[tensor_right_idx]
+        # given ordered anchor, pos, neg,
+        # assure first two labels are identical, first third label is different
+        assert tensor_left_label == tensor_middle_label
+        assert tensor_left_label != tensor_right_label
+
+
+@pytest.mark.parametrize('cut_index', [0, 1])
+def test_triplet_miner_given_insufficient_inputs(
+    embeddings, labels, siamese_miner, cut_index
+):
+    embeddings = embeddings[:cut_index]
+    labels = labels[:cut_index]
+    rv = list(siamese_miner.mine(embeddings, labels))
+    assert len(rv) == 0