aboutsummaryrefslogtreecommitdiff
path: root/kmeans.py
blob: a88e7737d0976b39c04945180aa6bbef3aa009e4 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
#!/usr/bin/env python3
"""Simple k-means implementation"""
import argparse
import math
import random


def distance(p1, p2) -> float:
    """return the euclidean distance between to points"""

    assert (len(p1) == len(p2))

    if isinstance(p1, tuple):
        p1 = list(p1)
    if isinstance(p2, tuple):
        p2 = list(p2)

    d = 0
    for i, x in enumerate(p1):
        d += (x - p2[i])**2

    return math.sqrt(d)


def gen_points(n, dimensions, max_value, min_value) -> list:
    """generate random points"""
    points = []
    for _ in range(n):
        point = []
        for _ in range(dimensions):
            point.append(random.randint(min_value, max_value))
        points.append(point)
    return points


def init_clusters(n, points) -> list:
    """create random initial clusters
    
    The clusters are chosen randomly between the encountered maximum and
    minimum values in the given points"""

    max_values = []
    min_values = []

    for i in range(len(points[0])):
        max_values.append(max(p[i] for p in points))
        min_values.append(min(p[i] for p in points))

    clusters = []
    for _ in range(n):
        cluster = []
        for i, max_value in enumerate(max_values):
            cluster.append(random.randint(min_values[i], max_value))
        clusters.append(tuple(cluster))

    return clusters


def map_points(clusters, points) -> dict:
    """map points to clusters

    Map each points to the cluster center with the minimal distance."""

    mapping = {p: [] for p in clusters}  # type: ignore
    for p in points:
        min_distance = None
        center = None
        for cluster in clusters:
            d = distance(p, cluster)
            if not min_distance or d < min_distance:
                center = cluster
                min_distance = d
        mapping[center].append(p)

    return mapping


def calc_clusters(mapping) -> list:
    """Calculate and return new centers for each cluster"""
    # return the new centers, use the old ones if no points are in the cluster
    return [
        k_mean(points) if len(points) else center
        for center, points in mapping.items()
    ]


def k_mean(points) -> tuple:
    """return the k-means of the given points"""
    # Start with the values of the first points
    means = list(points[0])

    # Only consider the remaining points
    for p in points[1:]:
        # calculate a running mean for each value
        for i, x in enumerate(p):
            means[i] = (means[i] + x) / 2

    return tuple(means)


def visualize(mapping):
    """visualize a mapping using matplotlib"""
    import matplotlib.pyplot as plt

    for center, points in mapping.items():
        x = [p[0] for p in points]
        y = [p[1] for p in points]
        plt.scatter(x, y, alpha=0.5)
        plt.plot(center[0], center[1], 'x')
    plt.show()


def main():
    """main entry point"""
    parser = argparse.ArgumentParser()
    parser.add_argument("-n", "--clusters", type=int, default=3)
    parser.add_argument("-r", "--random-points", type=int, default=10)
    parser.add_argument("-d", "--point-dimensions", type=int, default=2)
    parser.add_argument("--dont-visualize", action='store_true')
    parser.add_argument("--max", type=int, default=10)
    parser.add_argument("--min", type=int, default=1)
    args = parser.parse_args()

    points = gen_points(args.random_points, args.point_dimensions, args.max,
                        args.min)
    clusters = init_clusters(args.clusters, points)
    while True:
        mapping = map_points(clusters, points)

        if not args.dont_visualize:
            visualize(mapping)

        new_clusters = calc_clusters(mapping)

        if new_clusters == clusters:
            break
        clusters = new_clusters

    print(clusters)


if __name__ == '__main__':
    main()