SOM.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import numpy as np
import matplotlib.pyplot as plt
#-------------------------------------------------------------------------------
class Node(object):
    def __init__(self, x, y, input_length):
        self.x = x
        self.y = y
        self.weights = np.random.random(input_length)
        np.random.rand
        self.cluster = np.array([])

    def getWeights(self):
        return self.weights

    def updateWeights(self, alpha, influence, difference):
        self.weights += alpha * influence * difference

    def activate(self, input_vector):
        return np.sqrt(np.sum(((self.weights - input_vector) ** 2)))

    def getDistance(self, x, y):
        return np.sqrt((self.x - x) ** 2 + (self.y - y) ** 2)

    def getCluster(self):
        return self.cluster

    def setCluster(self):
        ones = (self.cluster == 1).sum()
        zeros = (self.cluster == 0).sum()
        if (ones > zeros):
            self.cluster = 1
        elif (ones == zeros):
            self.cluster = np.random.randint(0, 2)
        else:
            self.cluster = 0

#-------------------------------------------------------------------------------
class SOM:
    def __init__(self, map_size, input_length, clusters, alpha_start, alpha_min, decay_rate):
        self.map_size = map_size
        self.input_length = input_length
        self.clusters = np.zeros(clusters)
        self.alpha = alpha_start
        self.alpha_min = alpha_min
        self.decay_rate = decay_rate
        # Generowanie siatki neuronów
        self.neurons = np.array([[Node(x, y, self.input_length) for y in range(self.map_size)] for x in range(self.map_size)])
        # Generowanie macierzy odległości
        self.distances = np.zeros([self.map_size, self.map_size])
        self.neighbourhood_radius = map_size  # Promień sąsiedztwa. Tyle, bo wujek Kohonen tak zalecał.
        self.t = 0  # iterator

    def activate(self, input_vector):
        for x in range(self.map_size):
            for y in range(self.map_size):
                self.distances[x][y] = self.neurons[x][y].activate(input_vector)
    def adaptLearningRate(self):
        self.alpha *= np.exp(-self.t / self.decay_rate)

    def calculateNeighbourhoodRadius(self):
        self.neighbourhood_radius = self.map_size * np.exp(-self.t / self.decay_rate)

    def calculateInfluence(self, distance):
        return np.exp(-distance ** 2 / (2 * self.neighbourhood_radius ** 2))

    def train(self, input_vector, clusters_vector):
        self.t = 0
        # Łączna liczba elementów zbioru uczącego,
        # Czyli Liczba_wektorów x wymiar_wektora
        size = np.size(input_vector)
        vectors_num = size // self.input_length  # Liczba wektorów zbioru uczącego
        while (self.alpha > self.alpha_min):
            # Wybór losowego wektora
            i = np.random.randint(0, vectors_num)
            self.activate(input_vector[i])
            winner_x, winner_y = np.unravel_index(self.distances.argmin(), self.distances.shape)
            # Akutalizacja współczynników sieci
            self.adaptLearningRate()
            self.calculateNeighbourhoodRadius()
            for x in range(self.map_size):
                for y in range(self.map_size):
                    # Dla każdego węzła w sieci wyznacz jego odległość od węzła zwycięskiego
                    distance = self.neurons[x][y].getDistance(winner_x, winner_y)
                    if(distance <= self.neighbourhood_radius):
                        influence = self.calculateInfluence(distance)
                        difference = (input_vector[i] - self.neurons[x][y].getWeights())
                        self.neurons[x][y].updateWeights(self.alpha, influence, difference)
                        self.neurons[x][y].cluster = np.append(self.neurons[x][y].cluster, clusters_vector[i])
            self.t += 1
        for x in range(self.map_size):
                for y in range(self.map_size):
                    self.neurons[x][y].setCluster()

    def test(self, input_vector, clusters_vector):
        results = []
        (vectors_num, vector_length) = input_vector.shape
        for i in range(vectors_num):
            self.activate(input_vector[i])
            winner = np.unravel_index(self.distances.argmin(), self.distances.shape)
            results.append(self.neurons[winner[0]][winner[1]].getCluster())
        results = np.array(results)
        accuracy = 1 - np.sum(np.abs(results - clusters_vector)) / len(results)
        return accuracy

#-------------------------------------------------------------------------------
if __name__ == '__main__':
    map_size = 20
    clusters = 2
    input_length = 22
    decay_rate = 5
    alpha_min = 0.001
    alpha_start = 1

    filename = "Data/SPECT.train"
    data = np.loadtxt(filename, 'i8', delimiter = ',')
    train_features = (data.T[1:]).T
    train_overall_diagnosis = (data.T[:1][0])

    filename = "Data/SPECT.test"
    data = np.loadtxt(filename, 'i8', delimiter = ',')
    test_features = (data.T[1:]).T
    test_overall_diagnosis = (data.T[:1]).T

    som = SOM(map_size, input_length, clusters, alpha_start, alpha_min, decay_rate)
    som.train(train_features, train_overall_diagnosis)
    print(som.test(train_features, train_overall_diagnosis))