<commit_before># ----------------------------------------------------------------------
# Numenta Platform for Intelligent Computing (NuPIC)
# Copyright (C) 2016, Numenta, Inc.  Unless you have purchased from
# Numenta, Inc. a separate commercial license for this software code, the
# following terms and conditions apply:
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
# See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see http://www.gnu.org/licenses.
#
# http://numenta.org/licenses/
# ----------------------------------------------------------------------

import numpy
import os
import shutil
import tempfile
import unittest

from htmresearch.encoders import EncoderTypes
from htmresearch.frameworks.nlp.imbu import ImbuModels
from htmresearch.frameworks.nlp.classification_model import (
  ClassificationModel
)
from htmresearch.frameworks.nlp.classify_network_api import (
  ClassificationNetworkAPI
)



TEST_DATA_DIR = os.path.abspath(
  os.path.join(
    os.path.dirname(
      os.path.realpath(__file__)
    ),
    "..", "..", "nlp", "unit", "test_data"
  )
)



class TestImbu(unittest.TestCase):

  def setUp(self):
    self.dataPath = os.path.join(TEST_DATA_DIR, "sample_reviews_subset.csv")


  def tearDown(self):
    if os.path.exists("fake_cache_root"):
      shutil.rmtree("fake_cache_root")


  def _setupFakeImbuModelsInstance(self):
    return ImbuModels(
      cacheRoot="fake_cache_root",
      dataPath=self.dataPath,
      retina="en_associative",
      apiKey=os.environ.get("CORTICAL_API_KEY")
    )


  def _createTempModelCheckpoint(self):
    tmpDir = tempfile.mkdtemp()
    self.addCleanup(shutil.rmtree, tmpDir)
    return os.path.join(tmpDir, "checkpoint")


  def testCreateModel(self):
    imbu = self._setupFakeImbuModelsInstance()

    # You must specify a known model type
    self.assertRaises(TypeError, imbu.createModel, "random model type")

    # Assert that you can create models of known types from scratch (i.e. with
    # empty loadPath value)
    for modelType in imbu.modelMappings:
      # You _must_ specify loadPath and savePath.  Assert that attempts to not
      # specify will result in failure
      self.assertRaises(TypeError, imbu.createModel, modelType)
      self.assertRaises(TypeError, imbu.createModel, modelType, None)
      self.assertRaises(ValueError, imbu.createModel, modelType, None, None)

      # Attempt to create model using default arguments
      model = imbu.createModel(
        modelType, "", "", networkConfigName="imbu_sensor_knn.json"
      )

      # Assert model created was one of the expected types
      self.assertTrue(isinstance(model, ClassificationModel) or
                      isinstance(model, ClassificationNetworkAPI))


  def _exerciseModelLifecycle(self, modelType, queryTerm="unicorn",
                              networkConfigName="imbu_sensor_knn.json"):
    imbu = self._setupFakeImbuModelsInstance()

    checkpointLocation = self._createTempModelCheckpoint()

    # Create model with no load path.  This should trigger the creation of a
    # new intance, and train it.  Specify save path so that the trained model
    # gets saved
    originalModel = imbu.createModel(modelType,
                                     loadPath="",
                                     savePath=checkpointLocation,
                                     networkConfigName=networkConfigName)

    # Make a query, keep around for later comparison
    originalQueryResults = imbu.query(originalModel, queryTerm)

    del originalModel

    # Create model, specifying previous load path
    model = imbu.createModel(modelType,
                             loadPath=checkpointLocation,
                             savePath=checkpointLocation,
                             networkConfigName=networkConfigName)

    self.assertTrue(all(numpy.array_equal(original, new)
                        for (original, new)
                        in zip(originalQueryResults,
                               imbu.query(model, queryTerm))))


  def testCreateSaveLoadCioWordFingerprintModel(self):
    self._exerciseModelLifecycle("CioWordFingerprint")


  def testCreateSaveLoadCioDocumentFingerprintModel(self):
    self._exerciseModelLifecycle("CioDocumentFingerprint")


  def testCreateSaveLoadHTMNetworkModel(self):
    self._exerciseModelLifecycle("HTMNetwork")


  def testLoadKeywordsModel(self):
    self._exerciseModelLifecycle("Keywords")


  def testMappingModelNamesToModelTypes(self):
    imbu = ImbuModels(dataPath=self.dataPath)

    for modelName, mappedName in imbu.modelMappings.iteritems():
      self.assertEquals(mappedName, imbu._mapModelName(modelName),
        "Incorrect mapping returned for model named '{}'".format(modelName))

    self.assertRaises(ValueError, imbu._mapModelName, "fakeModel")


  def _checkNLPObjectParams(self, nlpObject, paramsToCheck):
    for key, value in paramsToCheck.iteritems():
      param = getattr(nlpObject, key)
      self.assertEquals(value, param,
        "The {} param for {} is not as expected.".format(key, repr(nlpObject)))


  def testSetParamsInModelFactory(self):
    imbu = self._setupFakeImbuModelsInstance()

    checkpointLocation = self._createTempModelCheckpoint()

    # Base set of required Cio params to check (in the encoder)
    cacheRoot = "fake_cache_root"
    paramsToCheck = dict(
      retina="en_associative",
      apiKey=os.environ.get("CORTICAL_API_KEY"),
      retinaScaling=1.0,
    )<commit_msg>unit test lifecycle for all Imbu models<commit_after># ----------------------------------------------------------------------
# Numenta Platform for Intelligent Computing (NuPIC)
# Copyright (C) 2016, Numenta, Inc.  Unless you have purchased from
# Numenta, Inc. a separate commercial license for this software code, the
# following terms and conditions apply:
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
# See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see http://www.gnu.org/licenses.
#
# http://numenta.org/licenses/
# ----------------------------------------------------------------------

import numpy
import os
import shutil
import tempfile
import unittest

from htmresearch.encoders import EncoderTypes
from htmresearch.frameworks.nlp.imbu import ImbuModels
from htmresearch.frameworks.nlp.classification_model import (
  ClassificationModel
)
from htmresearch.frameworks.nlp.classify_network_api import (
  ClassificationNetworkAPI
)



TEST_DATA_DIR = os.path.abspath(
  os.path.join(
    os.path.dirname(
      os.path.realpath(__file__)
    ),
    "..", "..", "nlp", "unit", "test_data"
  )
)



class TestImbu(unittest.TestCase):

  def setUp(self):
    self.dataPath = os.path.join(TEST_DATA_DIR, "sample_reviews_subset.csv")


  def tearDown(self):
    if os.path.exists("fake_cache_root"):
      shutil.rmtree("fake_cache_root")


  def _setupFakeImbuModelsInstance(self, retina="en_associative"):
    return ImbuModels(
      cacheRoot="fake_cache_root",
      dataPath=self.dataPath,
      retina=retina,
      apiKey=os.environ.get("CORTICAL_API_KEY")
    )


  def _createTempModelCheckpoint(self):
    tmpDir = tempfile.mkdtemp()
    self.addCleanup(shutil.rmtree, tmpDir)
    return os.path.join(tmpDir, "checkpoint")


  def testCreateModel(self):
    imbu = self._setupFakeImbuModelsInstance()

    # You must specify a known model type
    self.assertRaises(TypeError, imbu.createModel, "random model type")

    # Assert that you can create models of known types from scratch (i.e. with
    # empty loadPath value)
    for modelType in imbu.modelMappings:
      # You _must_ specify loadPath and savePath.  Assert that attempts to not
      # specify will result in failure
      self.assertRaises(TypeError, imbu.createModel, modelType)
      self.assertRaises(TypeError, imbu.createModel, modelType, None)
      self.assertRaises(ValueError, imbu.createModel, modelType, None, None)

      # Attempt to create model using default arguments
      model = imbu.createModel(
        modelType, "", "", networkConfigName="imbu_sensor_knn.json"
      )

      # Assert model created was one of the expected types
      self.assertTrue(isinstance(model, ClassificationModel) or
                      isinstance(model, ClassificationNetworkAPI))


  def _exerciseModelLifecycle(self, modelType, queryTerm="unicorn",
                              networkConfigName="imbu_sensor_knn.json",
                              retina="en_associative"):
    """ Create, save, load, and assert consistent results."""

    imbu = self._setupFakeImbuModelsInstance(retina=retina)

    checkpointLocation = self._createTempModelCheckpoint()

    # Create model with no load path.  This should trigger the creation of a
    # new intance, and train it.  Specify save path so that the trained model
    # gets saved
    originalModel = imbu.createModel(modelType,
                                     loadPath="",
                                     savePath=checkpointLocation,
                                     networkConfigName=networkConfigName)

    # Make a query, keep around for later comparison
    originalQueryResults = imbu.query(originalModel, queryTerm)

    del originalModel

    # Create model, specifying previous load path
    model = imbu.createModel(modelType,
                             loadPath=checkpointLocation,
                             savePath=checkpointLocation,
                             networkConfigName=networkConfigName)

    self.assertTrue(all(numpy.array_equal(original, new)
                        for (original, new)
                        in zip(originalQueryResults,
                               imbu.query(model, queryTerm))))


  def testCreateSaveLoadCioWordFingerprint(self):
    self._exerciseModelLifecycle("CioWordFingerprint")


  def testCreateSaveLoadCioDocumentFingerprint(self):
    self._exerciseModelLifecycle("CioDocumentFingerprint")


  def testCreateSaveLoadKeywords(self):
    self._exerciseModelLifecycle("Keywords")


  def testCreateSaveLoadSensorNetwork(self):
    self._exerciseModelLifecycle(
      "HTM_sensor_knn",
      networkConfigName="imbu_sensor_knn.json")


  def testCreateSaveLoadSensorSimpleTPNetwork(self):
    self._exerciseModelLifecycle(
      "HTM_sensor_simple_tp_knn",
      networkConfigName="imbu_sensor_simple_tp_knn.json",
      retina="en_associative_64_univ")


  def testCreateSaveLoadSensorTMSimpleTPNetwork(self):
    self._exerciseModelLifecycle(
      "HTM_sensor_tm_simple_tp_knn",
      networkConfigName="imbu_sensor_tm_simple_tp_knn.json",
      retina="en_associative_64_univ")


  def testMappingModelNamesToModelTypes(self):
    imbu = ImbuModels(dataPath=self.dataPath)

    for modelName, mappedName in imbu.modelMappings.iteritems():
      self.assertEquals(mappedName, imbu._mapModelName(modelName),
        "Incorrect mapping returned for model named '{}'".format(modelName))

    self.assertRaises(ValueError, imbu._mapModelName, "fakeModel")


  def _checkNLPObjectParams(self, nlpObject, paramsToCheck):
    for key, value in paramsToCheck.iteritems():
      param = getattr(nlpObject, key)
      self.assertEquals(value, param,
        "The {} param for {} is not as expected.".format(key, repr(nlpObject)))


  def testSetParamsInModelFactory(self):
    imbu = self._setupFakeImbuModelsInstance()

    checkpointLocation = self._createTempModelCheckpoint()

    # Base set of required Cio params to check (in the encoder)
    cacheRoot = "fake_cache_root"
    paramsToCheck = dict(
      retina="en_associative",
      apiKey=os.environ.get("CORTICAL_API_KEY"),
      retinaScaling=1.0,
    )
{'fraction_non_alphanumeric': 0.10221437740693197, 'fraction_numerical': 0.0014441591784338896, 'mean_word_length': 3.0365932642487046, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7373531', 'n_tokens_mistral': 3583, 'n_tokens_neox': 3438, 'n_words': 885}