Skip to content

K-Means

KMeansCorrespondenceSelector(k, feature_extractor)

Bases: CorrespondenceSelectorBase

K-means correspondence selector.

Initialize KMeansCorrespondenceSelector.

Parameters:

Name Type Description Default
k int

Number of clusters.

 required
feature_extractor FeatureExtractor

Feature extractor to use.

 required
Source code in src/compas_camcal/correspondence/selection/kmeans.py 
16
17
18
19
20
21
22
23
24
25
26
27
def __init__(self, k: int, feature_extractor: FeatureExtractor):
    """
    Initialize KMeansCorrespondenceSelector.

    Args:
        k: Number of clusters.
        feature_extractor: Feature extractor to use.
    """
    super().__init__()

    self.k = k
    self.feature_extractor = feature_extractor

select(correspondences)

Select a subset of correspondences from a collection of correspondences.

Parameters:

Name Type Description Default
correspondences Iterable[Correspondence]

Collection of correspondences to select from.

 required

Returns:

Type Description
Iterable[Correspondence]

Selected correspondences.
Source code in src/compas_camcal/correspondence/selection/kmeans.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def select(self, correspondences: Iterable[Correspondence]) -> Iterable[Correspondence]:
    """
    Select a subset of correspondences from a collection of correspondences.

    Args:
        correspondences: Collection of correspondences to select from.

    Returns:
        Selected correspondences.
    """
    features = np.array(
        [
            self.feature_extractor(correspondence.image_points[0])
            for correspondence in correspondences
        ]
    )
    self.logger.info(f"Extracted {features.shape[0]} features")

    # Find k clusters
    cluster_finder = ClusterFinder(features, self.k)
    self.logger.info("Computed clusters")

    # Group correspondences by cluster
    groups = {}
    for cluster_label, correspondence in zip(cluster_finder.Y_train, correspondences):
        if cluster_label not in groups:
            groups[cluster_label] = []
        groups[cluster_label].append(correspondence)
    correspondence_distribution = {
        cluster_label: len(correspondences)
        for cluster_label, correspondences in sorted(groups.items())
    }
    self.logger.debug(f"Grouped correspondence distribution: {correspondence_distribution}")

    # Select first correspondence in each group
    return [group[0] for group in groups.values()]