Notebooks/GhosalHTM.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "5903672a-e3db-4ee6-9433-0993c5382a2d",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os \n",
    "#import simpy\n",
    "import random\n",
    "import numpy as np\n",
    "#import networkx as nx\n",
    "#import matplotlib\n",
    "#matplotlib.use('Gtk3Agg')\n",
    "#import matplotlib.pyplot as plty\n",
    "import collections\n",
    "import re \n",
    "import string\n",
    "import os \n",
    "import pickle"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "22c47561-45bb-4f2e-9927-49f01755dc8b",
   "metadata": {},
   "outputs": [],
   "source": [
    "class G:\n",
    "    # WORKING_DIRECTORY = \"/Users/dghosal/Documents/HTM-Work\"\n",
    "    WORKING_DIRECTORY = os.getcwd()\n",
    "\n",
    "    # Network paramaters\n",
    "    number_of_columns = 1024\n",
    "    cells_per_column = 16\n",
    "    \n",
    "    \n",
    "    list_of_column_ids = list(range(1,number_of_columns+1))\n",
    "    list_of_ids_in_a_column = list(range(1,cells_per_column+1))\n",
    "    \n",
    "    # cell_id = (colum_id, cell_in_column_id)\n",
    "    \n",
    "    \n",
    "    # Cell parameters\n",
    "    number_of_segments_min = 64\n",
    "    number_of_segments_max = 64\n",
    "    fixed_number_of_segments = 64\n",
    "    \n",
    "    \n",
    "    # segment parameters\n",
    "    max_num_synapses = 30\n",
    "    min_num_synapses = 20\n",
    "    fixed_num_synapses = 25\n",
    "    \n",
    "    theta = 16\n",
    "    \n",
    "     \n",
    "    # synapse parameter\n",
    "    \n",
    "    max_p_value = 1\n",
    "    min_p_value = 0\n",
    "\n",
    "    alpha = 0.3 \n",
    "    \n",
    "    \n",
    "    # synaptic strength update parameters \n",
    "    \n",
    "    small_decrement = -0.05\n",
    "    very_small_decrement = -0.01\n",
    "    large_increment = 0.36\n",
    "    \n",
    "    \n",
    "    # encoding parameters  \n",
    "    encoding_length = 32  # number of columns activated for every input. \n",
    "    encoding_range = number_of_columns # \n",
    "    \n",
    "    \n",
    "    # This file will contain the encodings\n",
    "    encodings_filename = \"encodings_file.txt\"\n",
    "    \n",
    "    \n",
    "    \n",
    "    # epoch length\n",
    "    \n",
    "    epoch_length = 12\n",
    "    \n",
    "    # training file\n",
    "    pre_training_filename = \"tkamb_training.txt\"\n",
    "    #training_filename = \"tkamb_training.txt\"\n",
    "    #training_filename_long = \"tkamb_training_long.txt\"\n",
    "    #training_filename = \"tkamb.txt\"\n",
    "    \n",
    "    #training_filename = \"training_file-two-inputs.txt\"\n",
    "\n",
    "    training_filename = \"mapped_data_22093.txt\"\n",
    "    \n",
    "    # testing file\n",
    "    testing_filename = \"testing_file.txt\"\n",
    "    \n",
    "    # output file\n",
    "    output_filename = \"output_file_v4_sm00_22093.txt\"\n",
    "    \n",
    "    # pickle file to store the network\n",
    "    pickle_filename = \"my_network_v4_sm00.p\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "a1e1901f-0a15-4499-bc3d-f74ca37c14a6",
   "metadata": {},
   "outputs": [],
   "source": [
    "class sparse_encoder_decoder(object):\n",
    "    \n",
    "    def __init__(self, encoding_length):\n",
    "        \n",
    "        self.encoding_length = encoding_length\n",
    "        self.encoding_range = G.encoding_range\n",
    "        self.encodings = {}  #his is a dictionary of the encodings \n",
    "        self.population = list(range(1, self.encoding_range +1))\n",
    "    \n",
    "    def sparse_encode(self, input_char):\n",
    "        \n",
    "        if input_char in self.encodings.keys():\n",
    "            return(self.encodings[input_char])\n",
    "        else:\n",
    "            # give an input char it will generate an encoding and store the encodings \n",
    "            # the dictionary \n",
    "            # We will assume that we will prepopulate this dictionary\n",
    "            # First generater sample of length K \n",
    "            \n",
    "            temp = random.sample(self.population, self.encoding_length)\n",
    "            e = set(sorted(temp))\n",
    "        \n",
    "            while e in self.encodings.values():\n",
    "                temp = random.sample(self.population, self.encoding_length)\n",
    "                e = set(sorted(temp))\n",
    "            \n",
    "            self.encodings.update({input_char:e})\n",
    "            return(e)\n",
    "\n",
    "    def sparse_decode(self, input_char):\n",
    "        # given an input_char retiurs the tuple\n",
    "        return(self.encodings[input_char])\n",
    "    \n",
    "  \n",
    "    def load_current_encodings_from_dictionary(self, filename):\n",
    "        file = open(filename, \"r\")\n",
    "        lines = file.readlines()\n",
    "        for line in lines: \n",
    "            c = line.split(\":\")\n",
    "            key = c[0]\n",
    "            value = eval(c[1])\n",
    "            self.encodings.update({key:value})\n",
    "        file.close()\n",
    "        \n",
    "    def store_current_encoding_in_dictionary(self,filename):\n",
    "        file = open(filename, \"w\")\n",
    "        for key in self.encodings.keys():\n",
    "            line = str(key) + \":\" + str(self.encodings[key]) + \"\\n\"\n",
    "            file.write(line)\n",
    "        file.close()\n",
    "        \n",
    "    def generate_encodings(self, input_filename):\n",
    "        \n",
    "        # takes a filenname which is plain text \n",
    "        # removes all spaces and punctuations and get all the words and then the characters\n",
    "        # generates the encodings\n",
    "        inputfile = open(input_filename, 'r')\n",
    "        lines = inputfile.readlines()\n",
    "        for line in lines:\n",
    "            words = re.findall(r'\\w+', line) \n",
    "            #print(words)\n",
    "            for word in words:\n",
    "                for character in list(word):\n",
    "                    self.sparse_encode(character.lower()) \n",
    "        \n",
    "        inputfile.close()\n",
    "\n",
    "        \n",
    "   \n",
    "    def print_encodings(self):\n",
    "        \n",
    "        # print the encoded dioctionary\n",
    "        for key in self.encodings.keys():\n",
    "            print(key, self.encodings[key])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "63589d65-778a-4f8a-b63b-0f30ffb6b5da",
   "metadata": {},
   "outputs": [],
   "source": [
    "class synapse(object):\n",
    "    \n",
    "    def __init__(self, synapse_id, segment_id, local_cell_id): \n",
    "        self.id = synapse_id                           # an integer\n",
    "        self.segment_id = segment_id                   # which segment does the synapse belong to\n",
    "        self.local_cell_id = local_cell_id                   # which cell does it belong to. This is a tuple\n",
    "        self.remote_cell_id = ()                       # which cell does it connect to \n",
    "                                                       #   we assume each synapse connect to a single cell which is a tuple\n",
    "        self.p_value = np.random.uniform(0,1)          # the permanance value which is initialized to a value between 0 and 1\n",
    "        \n",
    "    \n",
    "    def get_p_value(self):\n",
    "        return(self.p_value)\n",
    "    \n",
    "    def update_p_value(self, p_value_change):\n",
    "        old_p_value = self.p_value\n",
    "        self.p_value = min(max(self.p_value + p_value_change,0),1)\n",
    "        #print(\"Updated synapse {0} in segment {1} in cell {2}  with remote cell_id {3} with current p_Value {4} updated to new p_value {5}\".format(self.id, self.segment_id, self.local_cell_id, self.remote_cell_id, old_p_value, self.p_value))\n",
    "        \n",
    "    \n",
    "    def get_id(self): \n",
    "        return(self.id)\n",
    "        \n",
    "    def get_segment_id(self):\n",
    "        return(self.segment_id)\n",
    "    \n",
    "    def get_local_cell_id(self):                       # which cell does this synapse belong to\n",
    "        return(self.cell_id)\n",
    "    \n",
    "    def assign_remote_cell_id(self, remote_cell_id):\n",
    "        self.remote_cell_id = remote_cell_id\n",
    "    \n",
    "    def get_remote_cell_id(self):             # Which cell does this connect to \n",
    "        return(self.remote_cell_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "e7116d9b-a189-4b32-a067-f4b756aecbe9",
   "metadata": {},
   "outputs": [],
   "source": [
    "class segment(object):\n",
    "    \n",
    "    def __init__(self, segment_id, cell_id, num_synapses):\n",
    "        self.segment_id  = segment_id                # an integer id\n",
    "        self.local_cell_id = cell_id               # which cell does this segmeent belong to\n",
    "        self.num_synapses = num_synapses     # number of synapses\n",
    "        self.segment_synapses = {}           # dictionary of synapses of this segment. key is the synapse id  value is the synapse\n",
    "        self.dij = {}                        # dictionary key synapseid value is the permanence value\n",
    "        self.remote_cell_ids = set()            # a set containing all the remote cell ids of the segment\n",
    "        self.tildedij = {}                    # dictionary key synapseid value remote-cell-id for synapses with p_value > G.alpha\n",
    "        self.tildedij_set = set()                 # This is just the set consisting of the remote cellids of connected synapse\n",
    "        self.dotdij = {}                     # dictionary key synapseid value remote-cell-id for synapses with p_value > 0\n",
    "        self.dotdij_set = set()                  # This is just the set consisting of the remote cellids of synapses with +ve p_value\n",
    "    \n",
    "    def create_synapses(self): \n",
    "        \n",
    "        # this function will create the synapses \n",
    "        # initilize dij, tildedij, and dotdij the dictionaries and the sets\n",
    "                \n",
    "        for i in list(range(1,self.num_synapses+1)):\n",
    "            \n",
    "            # create the synapse\n",
    "            s = synapse(i, self.segment_id, self.local_cell_id)\n",
    "            \n",
    "            # store the synapse\n",
    "            self.segment_synapses[i] = s\n",
    "            \n",
    "            # connect the synapse to a remote cell\n",
    "            # generate a random cell_id \n",
    "            # we want to makem sure that all the synapses of the segment connect to \n",
    "            #   distinct cells \n",
    "\n",
    "            x = np.random.randint(1, G.number_of_columns+1)\n",
    "            y = np.random.randint(1, G.cells_per_column+1)\n",
    "            remote_cell_id = (x,y)\n",
    "            while (remote_cell_id in self.remote_cell_ids) or (remote_cell_id == self.local_cell_id): \n",
    "                x = np.random.randint(1, G.number_of_columns+1)\n",
    "                y = np.random.randint(1, G.cells_per_column+1)\n",
    "                remote_cell_id = (x,y)\n",
    "            \n",
    "            #print(remote_cell_id)\n",
    "            # assign the remoote cell-id to the synapse\n",
    "            s.assign_remote_cell_id(remote_cell_id)\n",
    "            \n",
    "            p_value = s.get_p_value()\n",
    "            \n",
    "            # add it to dij \n",
    "            #k = str(i) + ':' + str(remote_cell_id)\n",
    "            self.dij.update({i:p_value})\n",
    "            \n",
    "            # add the remote cell_id to the set self.remote_cell_ids\n",
    "            self.remote_cell_ids.add(remote_cell_id) \n",
    "            \n",
    "            # add to dotdij \n",
    "            if p_value > 0:\n",
    "                self.dotdij.update({i:remote_cell_id})\n",
    "                self.dotdij_set.add(remote_cell_id)\n",
    "\n",
    "            \n",
    "            # add to tildedij \n",
    "            if p_value > G.alpha:\n",
    "                self.tildedij.update({i:remote_cell_id})\n",
    "                self.tildedij_set.add(remote_cell_id)\n",
    " "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "13004a81-46c1-49c3-9daa-fece09922e2d",
   "metadata": {},
   "outputs": [],
   "source": [
    "class cell(object):\n",
    "    \n",
    "    def __init__(self, cell_id, num_segments):\n",
    "        self.cell_id = cell_id                 # this is a tuple\n",
    "        self.num_segments = num_segments       # how many segments does the cell have\n",
    "        self.cell_segments = {}                # dictionary of all the segments in the cell\n",
    "                                               #   the segment-id is the key and segment object is the value \n",
    "\n",
    "        self.activity_state = 0\n",
    "        self.predictive_state = 0\n",
    "        \n",
    "        self.active_segments = []             # this is a set of segment ids that have been found to be active                                   # \n",
    "        self.segment_strengths = {}                                # create dictionary segment_id: strength wrt to at\n",
    "\n",
    "        \n",
    "    def create_segments(self):\n",
    "        \n",
    "        for i in list(range(1, self.num_segments+1)):\n",
    "            #num_synapses = np.random.randint(G.min_num_synapses, G.max_num_synapses)   \n",
    "            num_synapses = G.fixed_num_synapses  \n",
    "            s = segment(i, self.cell_id, num_synapses)\n",
    "            s.create_synapses()\n",
    "            self.cell_segments[i] = s\n",
    "\n",
    "            \n",
    "            \n",
    "    def get_activity_state(self):\n",
    "        \n",
    "        return(self.activity_state)\n",
    "        \n",
    "        \n",
    "    def get_predictive_state(self):\n",
    "        \n",
    "        return(self.predictive_state)\n",
    "        \n",
    "        \n",
    "    def update_activity_state(self, value):\n",
    "        \n",
    "        # recompute the activity state of the cell   \n",
    "        \n",
    "        self.activity_state = value\n",
    "              \n",
    "\n",
    "    def compute_predictive_state(self, at):\n",
    "        \n",
    "        # this function will compute the predictive state given the currrent feedforward input \n",
    "        # We will break as soon as we find the first active segment\n",
    "        \n",
    "        count = 0\n",
    "        \n",
    "        for segment_id in list(range(1, self.num_segments+1)):\n",
    "            \n",
    "            # for this segment find the overlapp with at \n",
    "            overlapp = self.cell_segments[segment_id].compute_overlapp_with_tidledij(at)\n",
    "\n",
    "            if len(overlapp) > G.theta:\n",
    "                count += 1\n",
    "                break     \n",
    "            \n",
    "        if count == 0:\n",
    "            self.predictive_state = 0\n",
    "            return(0)\n",
    "        else:\n",
    "            self.predictive_state = 1\n",
    "            return(1)\n",
    "    \n",
    "    \n",
    "    \n",
    "    def determine_segment_id_causing_predictive_state(self, at):\n",
    "        \n",
    "        # given a set of cell ids in at\n",
    "        # generates a dictionary of  segment_id and strengths that caused this cell to be in the predictive state  \n",
    "        \n",
    "        self.segment_ids_causing_predictive_state = {}\n",
    "        for segment_id in self.cell_segments.keys(): \n",
    "            overlapp = self.cell_segments[segment_id].tildedij_set.intersection(at)\n",
    "            if len(overlapp) > G.theta:\n",
    "                self.segment_ids_causing_predictive_state.update({segment_id:len(overlapp)})\n",
    "        #print(\"Segments causing predictive state: {0}\".format(self.segment_ids_causing_predictive_state))\n",
    "        \n",
    "\n",
    "    def determine_segment_id_strengths(self, at):\n",
    "        \n",
    "        # given a set of cell ids in at\n",
    "        # returns a dictionaries of  segment_id and their overlapps\n",
    "        \n",
    "        self.segment_id_strengths = {}\n",
    "        \n",
    "        for segment_id in self.cell_segments.keys(): \n",
    "            overlapp = self.cell_segments[segment_id].dotdij_set.intersection(at)\n",
    "            self.segment_id_strengths.update({segment_id:overlapp})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6cecff93-891b-4f6e-828a-874df703e5ee",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}