SpatialPooler/spatial_pooler.py at master · CSDUMMI/SpatialPooler · GitHub

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
#! /usr/bin/python3
import numpy as np
import sys, random, json


def overlap(x,y):
    # You can't overlap sdrs
    # with different shapes
    if x.shape != y.shape:
        return None
    result = np.zeros(x.shape,dtype=np.bool_)
    for i in range(x.shape[0]):
        result[i] = x[i] and y[i]
    return result


class SpatialPooler():

    def __init__(   self,
                    num_collumns=128,
                    input_size=256,
                    threshhold_permances=0.5,
                    threshhold_activation=0.6,
                    size_of_potential_pool=0.75,
                    permanence_inc=0.1,
                    permanence_dec=0.01):
        self.current = 0
        self.permanence_inc = permanence_inc
        self.permanence_dec = permanence_dec
        self.threshhold_permanences = threshhold_permances
        self.threshhold_activation = threshhold_activation
        self.collumns = self.init_collumn(num_collumns,input_size,1-size_of_potential_pool)


    def init_collumn(self,num_collumns,input_size,size_of_potential_pool=0.75):
        collumns = [{} for i in range(num_collumns)]
        for i in range(num_collumns):
            collumns[i]['potential_pool'] = np.random.rand(input_size) > size_of_potential_pool # Create a random potential pool with a certain percantage of potential connections
            collumns[i]['permanences'] = np.random.rand(np.sum(collumns[i]['potential_pool'])) # Random values for the permanence between  0-1
        return collumns


    def activation(self,state):
        """
        Sum state and then compare that to threshhold_activation
        """
        length = state.shape[0]
        return ((np.sum(state)/length) > self.threshhold_activation)

    def permanence(self,state):
        """
        For all permanences
        filter all those out that are below self.threshhold_permances
        And increase those that are above the same by self.permanence_inc
        """
        active = self.collumns[self.current]['permanences'] > self.threshhold_permanences
        inactive = self.collumns[self.current]['permanences'] <= self.threshhold_permanences
        self.collumns[self.current]['permanences'][active] += self.permanence_inc
        self.collumns[self.current]['permanences'][inactive] -= self.permanence_dec
        return (self.collumns[self.current]['permanences'] > self.threshhold_permanences) * state


    def potential_pool(self,state):
        predicat = self.collumns[self.current]['potential_pool']
        return np.extract(predicat,state)

    def spatial_pooler(self,state):
        """
        state := np.array (bool)
        Spatial Pooler:
        Input -> potetial pool -> permanence -> activation -> output
        """
        return self.activation(self.permanence(self.potential_pool(state)))

    def run(self,input_sdr):
        self.current = 0
        output = np.zeros(len(self.collumns),dtype=np.bool_)
        for i in range(len(self.collumns)):
            output[i] = self.spatial_pooler(input_sdr)
            self.current += 1
        self.current = 0
        return output