signature:
def _inhibitColumnsLocal ( self , overlaps , density ) :
    """Performs local inhibition . Local inhibition is performed on a column by
 column basis . Each column observes the overlaps of its neighbors and is
 selected if its overlap score is within the top ' numActive ' in its local
 neighborhood . At most half of the columns in a local neighborhood are
 allowed to be active . Columns with an overlap score below the
 ' stimulusThreshold ' are always inhibited .
 : param overlaps : an array containing the overlap score for each column .
 The overlap score for a column is defined as the number
 of synapses in a " connected state " ( connected synapses )
 that are connected to input bits which are turned on .
 : param density : The fraction of columns to survive inhibition . This
 value is only an intended target . Since the surviving
 columns are picked in a local fashion , the exact fraction
 of surviving columns is likely to vary .
 @ return list with indices of the winning columns"""

implementation:
activeArray = numpy . zeros ( self . _numColumns , dtype = "bool" )
for column , overlap in enumerate ( overlaps ) :
    if overlap >= self . _stimulusThreshold :
        neighborhood = self . _getColumnNeighborhood ( column )
        neighborhoodOverlaps = overlaps [ neighborhood ]
        numBigger = numpy . count_nonzero ( neighborhoodOverlaps > overlap )
        # When there is a tie , favor neighbors that are already selected as
        # active .
        ties = numpy . where ( neighborhoodOverlaps == overlap )
        tiedNeighbors = neighborhood [ ties ]
        numTiesLost = numpy . count_nonzero ( activeArray [ tiedNeighbors ] )
        numActive = int ( 0.5 + density * len ( neighborhood ) )
        if numBigger + numTiesLost < numActive :
            activeArray [ column ] = True
return activeArray . nonzero ( ) [ 0 ]


signature:
def _getColumnNeighborhood ( self , centerColumn ) :
    """Gets a neighborhood of columns .
 Simply calls topology . neighborhood or topology . wrappingNeighborhood
 A subclass can insert different topology behavior by overriding this method .
 : param centerColumn ( int )
 The center of the neighborhood .
 @ returns ( 1D numpy array of integers )
 The columns in the neighborhood ."""

implementation:
if self . _wrapAround :
    return topology . wrappingNeighborhood ( centerColumn , self . _inhibitionRadius , self . _columnDimensions )
else :
    return topology . neighborhood ( centerColumn , self . _inhibitionRadius , self . _columnDimensions )


signature:
def _getInputNeighborhood ( self , centerInput ) :
    """Gets a neighborhood of inputs .
 Simply calls topology . wrappingNeighborhood or topology . neighborhood .
 A subclass can insert different topology behavior by overriding this method .
 : param centerInput ( int )
 The center of the neighborhood .
 @ returns ( 1D numpy array of integers )
 The inputs in the neighborhood ."""

implementation:
if self . _wrapAround :
    return topology . wrappingNeighborhood ( centerInput , self . _potentialRadius , self . _inputDimensions )
else :
    return topology . neighborhood ( centerInput , self . _potentialRadius , self . _inputDimensions )


signature:
def _seed ( self , seed = - 1 ) :
    """Initialize the random seed"""

implementation:
if seed != - 1 :
    self . _random = NupicRandom ( seed )
else :
    self . _random = NupicRandom ( )


signature:
def handleGetValue ( self , topContainer ) :
    """This method overrides ValueGetterBase ' s " pure virtual " method . It
 returns the referenced value . The derived class is NOT responsible for
 fully resolving the reference ' d value in the event the value resolves to
 another ValueGetterBase - based instance - - this is handled automatically
 within ValueGetterBase implementation .
 topContainer : The top - level container ( dict , tuple , or list [ sub - ] instance )
 within whose context the value - getter is applied . If
 self . _ _ referenceDict is None , then topContainer will be used
 as the reference dictionary for resolving our dictionary key
 chain .
 Returns : The value referenced by this instance ( which may be another
 value - getter instance )"""

implementation:
value = self . __referenceDict if self . __referenceDict is not None else topContainer
for key in self . __dictKeyChain :
    value = value [ key ]
return value


signature:
def getInputNames ( self ) :
    """Returns list of input names in spec ."""

implementation:
inputs = self . getSpec ( ) . inputs
return [ inputs . getByIndex ( i ) [ 0 ] for i in xrange ( inputs . getCount ( ) ) ]


signature:
def getOutputNames ( self ) :
    """Returns list of output names in spec ."""

implementation:
outputs = self . getSpec ( ) . outputs
return [ outputs . getByIndex ( i ) [ 0 ] for i in xrange ( outputs . getCount ( ) ) ]


signature:
def _getParameterMethods ( self , paramName ) :
    """Returns functions to set / get the parameter . These are
 the strongly typed functions get / setParameterUInt32 , etc .
 The return value is a pair :
 setfunc , getfunc
 If the parameter is not available on this region , setfunc / getfunc
 are None ."""

implementation:
if paramName in self . _paramTypeCache :
    return self . _paramTypeCache [ paramName ]
try : # Catch the error here . We will re - throw in getParameter or
# setParameter with a better error message than we could generate here
    paramSpec = self . getSpec ( ) . parameters . getByName ( paramName )
except :
    return ( None , None )
dataType = paramSpec . dataType
dataTypeName = basicTypes [ dataType ]
count = paramSpec . count
if count == 1 : # Dynamically generate the proper typed get / setParameter < dataType >
    x = 'etParameter' + dataTypeName
    try :
        g = getattr ( self , 'g' + x )
        # get the typed getParameter method
        s = getattr ( self , 's' + x )
        # get the typed setParameter method
    except AttributeError :
        raise Exception ( "Internal error: unknown parameter type %s" % dataTypeName )
    info = ( s , g )
else :
    if dataTypeName == "Byte" :
        info = ( self . setParameterString , self . getParameterString )
    else :
        helper = _ArrayParameterHelper ( self , dataType )
        info = ( self . setParameterArray , helper . getParameterArray )
self . _paramTypeCache [ paramName ] = info
return info


signature:
def getParameter ( self , paramName ) :
    """Get parameter value"""

implementation:
( setter , getter ) = self . _getParameterMethods ( paramName )
if getter is None :
    import exceptions
    raise exceptions . Exception ( "getParameter -- parameter name '%s' does not exist in region %s of type %s" % ( paramName , self . name , self . type ) )
return getter ( paramName )


signature:
def setParameter ( self , paramName , value ) :
    """Set parameter value"""

implementation:
( setter , getter ) = self . _getParameterMethods ( paramName )
if setter is None :
    import exceptions
    raise exceptions . Exception ( "setParameter -- parameter name '%s' does not exist in region %s of type %s" % ( paramName , self . name , self . type ) )
setter ( paramName , value )


signature:
def getRegionsByType ( self , regionClass ) :
    """Gets all region instances of a given class
 ( for example , nupic . regions . sp _ region . SPRegion ) ."""

implementation:
regions = [ ]
for region in self . regions . values ( ) :
    if type ( region . getSelf ( ) ) is regionClass :
        regions . append ( region )
return regions


signature:
def getSpec ( cls ) :
    """Overrides : meth : ` nupic . bindings . regions . PyRegion . PyRegion . getSpec ` ."""

implementation:
ns = dict ( description = SDRClassifierRegion . __doc__ , singleNodeOnly = True , inputs = dict ( actValueIn = dict ( description = "Actual value of the field to predict. Only taken " "into account if the input has no category field." , dataType = "Real32" , count = 0 , required = False , regionLevel = False , isDefaultInput = False , requireSplitterMap = False ) , bucketIdxIn = dict ( description = "Active index of the encoder bucket for the " "actual value of the field to predict. Only taken " "into account if the input has no category field." , dataType = "UInt64" , count = 0 , required = False , regionLevel = False , isDefaultInput = False , requireSplitterMap = False ) , categoryIn = dict ( description = 'Vector of categories of the input sample' , dataType = 'Real32' , count = 0 , required = True , regionLevel = True , isDefaultInput = False , requireSplitterMap = False ) , bottomUpIn = dict ( description = 'Belief values over children\'s groups' , dataType = 'Real32' , count = 0 , required = True , regionLevel = False , isDefaultInput = True , requireSplitterMap = False ) , predictedActiveCells = dict ( description = "The cells that are active and predicted" , dataType = 'Real32' , count = 0 , required = True , regionLevel = True , isDefaultInput = False , requireSplitterMap = False ) , sequenceIdIn = dict ( description = "Sequence ID" , dataType = 'UInt64' , count = 1 , required = False , regionLevel = True , isDefaultInput = False , requireSplitterMap = False ) , ) , outputs = dict ( categoriesOut = dict ( description = 'Classification results' , dataType = 'Real32' , count = 0 , regionLevel = True , isDefaultOutput = False , requireSplitterMap = False ) , actualValues = dict ( description = 'Classification results' , dataType = 'Real32' , count = 0 , regionLevel = True , isDefaultOutput = False , requireSplitterMap = False ) , probabilities = dict ( description = 'Classification results' , dataType = 'Real32' , count = 0 , regionLevel = True , isDefaultOutput = False , requireSplitterMap = False ) , ) , parameters = dict ( learningMode = dict ( description = 'Boolean (0/1) indicating whether or not a region ' 'is in learning mode.' , dataType = 'UInt32' , count = 1 , constraints = 'bool' , defaultValue = 1 , accessMode = 'ReadWrite' ) , inferenceMode = dict ( description = 'Boolean (0/1) indicating whether or not a region ' 'is in inference mode.' , dataType = 'UInt32' , count = 1 , constraints = 'bool' , defaultValue = 0 , accessMode = 'ReadWrite' ) , maxCategoryCount = dict ( description = 'The maximal number of categories the ' 'classifier will distinguish between.' , dataType = 'UInt32' , required = True , count = 1 , constraints = '' , # arbitrarily large value
defaultValue = 2000 , accessMode = 'Create' ) , steps = dict ( description = 'Comma separated list of the desired steps of ' 'prediction that the classifier should learn' , dataType = "Byte" , count = 0 , constraints = '' , defaultValue = '0' , accessMode = 'Create' ) , alpha = dict ( description = 'The alpha is the learning rate of the classifier.' 'lower alpha results in longer term memory and slower ' 'learning' , dataType = "Real32" , count = 1 , constraints = '' , defaultValue = 0.001 , accessMode = 'Create' ) , implementation = dict ( description = 'The classifier implementation to use.' , accessMode = 'ReadWrite' , dataType = 'Byte' , count = 0 , constraints = 'enum: py, cpp' ) , verbosity = dict ( description = 'An integer that controls the verbosity level, ' '0 means no verbose output, increasing integers ' 'provide more verbosity.' , dataType = 'UInt32' , count = 1 , constraints = '' , defaultValue = 0 , accessMode = 'ReadWrite' ) , ) , commands = dict ( ) )
return ns


signature:
def initialize ( self ) :
    """Overrides : meth : ` nupic . bindings . regions . PyRegion . PyRegion . initialize ` .
 Is called once by NuPIC before the first call to compute ( ) .
 Initializes self . _ sdrClassifier if it is not already initialized ."""

implementation:
if self . _sdrClassifier is None :
    self . _sdrClassifier = SDRClassifierFactory . create ( steps = self . stepsList , alpha = self . alpha , verbosity = self . verbosity , implementation = self . implementation , )


signature:
def setParameter ( self , name , index , value ) :
    """Overrides : meth : ` nupic . bindings . regions . PyRegion . PyRegion . setParameter ` ."""

implementation:
if name == "learningMode" :
    self . learningMode = bool ( int ( value ) )
elif name == "inferenceMode" :
    self . inferenceMode = bool ( int ( value ) )
else :
    return PyRegion . setParameter ( self , name , index , value )


signature:
def writeToProto ( self , proto ) :
    """Write state to proto object .
 : param proto : SDRClassifierRegionProto capnproto object"""

implementation:
proto . implementation = self . implementation
proto . steps = self . steps
proto . alpha = self . alpha
proto . verbosity = self . verbosity
proto . maxCategoryCount = self . maxCategoryCount
proto . learningMode = self . learningMode
proto . inferenceMode = self . inferenceMode
proto . recordNum = self . recordNum
self . _sdrClassifier . write ( proto . sdrClassifier )


signature:
def readFromProto ( cls , proto ) :
    """Read state from proto object .
 : param proto : SDRClassifierRegionProto capnproto object"""

implementation:
instance = cls ( )
instance . implementation = proto . implementation
instance . steps = proto . steps
instance . stepsList = [ int ( i ) for i in proto . steps . split ( "," ) ]
instance . alpha = proto . alpha
instance . verbosity = proto . verbosity
instance . maxCategoryCount = proto . maxCategoryCount
instance . _sdrClassifier = SDRClassifierFactory . read ( proto )
instance . learningMode = proto . learningMode
instance . inferenceMode = proto . inferenceMode
instance . recordNum = proto . recordNum
return instance


signature:
def compute ( self , inputs , outputs ) :
    """Process one input sample .
 This method is called by the runtime engine .
 : param inputs : ( dict ) mapping region input names to numpy . array values
 : param outputs : ( dict ) mapping region output names to numpy . arrays that
 should be populated with output values by this method"""

implementation:
# This flag helps to prevent double - computation , in case the deprecated
# customCompute ( ) method is being called in addition to compute ( ) called
# when network . run ( ) is called
self . _computeFlag = True
patternNZ = inputs [ "bottomUpIn" ] . nonzero ( ) [ 0 ]
if self . learningMode : # An input can potentially belong to multiple categories .
# If a category value is < 0 , it means that the input does not belong to
# that category .
    categories = [ category for category in inputs [ "categoryIn" ] if category >= 0 ]
    if len ( categories ) > 0 : # Allow to train on multiple input categories .
        bucketIdxList = [ ]
        actValueList = [ ]
        for category in categories :
            bucketIdxList . append ( int ( category ) )
            if "actValueIn" not in inputs :
                actValueList . append ( int ( category ) )
            else :
                actValueList . append ( float ( inputs [ "actValueIn" ] ) )
        classificationIn = { "bucketIdx" : bucketIdxList , "actValue" : actValueList }
    else : # If the input does not belong to a category , i . e . len ( categories ) = = 0,
    # then look for bucketIdx and actValueIn .
        if "bucketIdxIn" not in inputs :
            raise KeyError ( "Network link missing: bucketIdxOut -> bucketIdxIn" )
        if "actValueIn" not in inputs :
            raise KeyError ( "Network link missing: actValueOut -> actValueIn" )
        classificationIn = { "bucketIdx" : int ( inputs [ "bucketIdxIn" ] ) , "actValue" : float ( inputs [ "actValueIn" ] ) }
else : # Use Dummy classification input , because this param is required even for
# inference mode . Because learning is off , the classifier is not learning
# this dummy input . Inference only here .
    classificationIn = { "actValue" : 0 , "bucketIdx" : 0 }
# Perform inference if self . inferenceMode is True
# Train classifier if self . learningMode is True
clResults = self . _sdrClassifier . compute ( recordNum = self . recordNum , patternNZ = patternNZ , classification = classificationIn , learn = self . learningMode , infer = self . inferenceMode )
# fill outputs with clResults
if clResults is not None and len ( clResults ) > 0 :
    outputs [ 'actualValues' ] [ : len ( clResults [ "actualValues" ] ) ] = clResults [ "actualValues" ]
    for step in self . stepsList :
        stepIndex = self . stepsList . index ( step )
        categoryOut = clResults [ "actualValues" ] [ clResults [ step ] . argmax ( ) ]
        outputs [ 'categoriesOut' ] [ stepIndex ] = categoryOut
        # Flatten the rest of the output . For example :
        # Original dict { 1 : [ 0.1 , 0.3 , 0.2 , 0.7]
        # 4 : [ 0.2 , 0.4 , 0.3 , 0.5 ] }
        # becomes : [ 0.1 , 0.3 , 0.2 , 0.7 , 0.2 , 0.4 , 0.3 , 0.5]
        stepProbabilities = clResults [ step ]
        for categoryIndex in xrange ( self . maxCategoryCount ) :
            flatIndex = categoryIndex + stepIndex * self . maxCategoryCount
            if categoryIndex < len ( stepProbabilities ) :
                outputs [ 'probabilities' ] [ flatIndex ] = stepProbabilities [ categoryIndex ]
            else :
                outputs [ 'probabilities' ] [ flatIndex ] = 0.0
self . recordNum += 1


signature:
def customCompute ( self , recordNum , patternNZ , classification ) :
    """Just return the inference value from one input sample . The actual
 learning happens in compute ( ) - - if , and only if learning is enabled - -
 which is called when you run the network .
 . . warning : : This method is deprecated and exists only to maintain backward
 compatibility . This method is deprecated , and will be removed . Use
 : meth : ` nupic . engine . Network . run ` instead , which will call
 : meth : ` ~ nupic . regions . sdr _ classifier _ region . compute ` .
 : param recordNum : ( int ) Record number of the input sample .
 : param patternNZ : ( list ) of the active indices from the output below
 : param classification : ( dict ) of the classification information :
 * ` ` bucketIdx ` ` : index of the encoder bucket
 * ` ` actValue ` ` : actual value going into the encoder
 : returns : ( dict ) containing inference results , one entry for each step in
 ` ` self . steps ` ` . The key is the number of steps , the value is an
 array containing the relative likelihood for each ` ` bucketIdx ` `
 starting from 0.
 For example :
 { ' actualValues ' : [ 0.0 , 1.0 , 2.0 , 3.0]
 1 : [ 0.1 , 0.3 , 0.2 , 0.7]
 4 : [ 0.2 , 0.4 , 0.3 , 0.5 ] }"""

implementation:
# If the compute flag has not been initialized ( for example if we
# restored a model from an old checkpoint ) initialize it to False .
if not hasattr ( self , "_computeFlag" ) :
    self . _computeFlag = False
if self . _computeFlag : # Will raise an exception if the deprecated method customCompute ( ) is
# being used at the same time as the compute function .
    warnings . simplefilter ( 'error' , DeprecationWarning )
    warnings . warn ( "The customCompute() method should not be " "called at the same time as the compute() " "method. The compute() method is called " "whenever network.run() is called." , DeprecationWarning )
return self . _sdrClassifier . compute ( recordNum , patternNZ , classification , self . learningMode , self . inferenceMode )


signature:
def getOutputElementCount ( self , outputName ) :
    """Overrides : meth : ` nupic . bindings . regions . PyRegion . PyRegion . getOutputElementCount ` ."""

implementation:
if outputName == "categoriesOut" :
    return len ( self . stepsList )
elif outputName == "probabilities" :
    return len ( self . stepsList ) * self . maxCategoryCount
elif outputName == "actualValues" :
    return self . maxCategoryCount
else :
    raise ValueError ( "Unknown output {}." . format ( outputName ) )


signature:
def run ( self ) :
    """Runs the OPF Model
 Parameters :
 retval : ( completionReason , completionMsg )
 where completionReason is one of the ClientJobsDAO . CMPL _ REASON _ XXX
 equates ."""

implementation:
# Load the experiment ' s description . py module
descriptionPyModule = helpers . loadExperimentDescriptionScriptFromDir ( self . _experimentDir )
expIface = helpers . getExperimentDescriptionInterfaceFromModule ( descriptionPyModule )
expIface . normalizeStreamSources ( )
modelDescription = expIface . getModelDescription ( )
self . _modelControl = expIface . getModelControl ( )
# Create the input data stream for this task
streamDef = self . _modelControl [ 'dataset' ]
from nupic . data . stream_reader import StreamReader
readTimeout = 0
self . _inputSource = StreamReader ( streamDef , isBlocking = False , maxTimeout = readTimeout )
# Get field statistics from the input source
fieldStats = self . _getFieldStats ( )
# Construct the model instance
self . _model = ModelFactory . create ( modelDescription )
self . _model . setFieldStatistics ( fieldStats )
self . _model . enableLearning ( )
self . _model . enableInference ( self . _modelControl . get ( "inferenceArgs" , None ) )
# Instantiate the metrics
self . __metricMgr = MetricsManager ( self . _modelControl . get ( 'metrics' , None ) , self . _model . getFieldInfo ( ) , self . _model . getInferenceType ( ) )
self . __loggedMetricPatterns = self . _modelControl . get ( "loggedMetrics" , [ ] )
self . _optimizedMetricLabel = self . __getOptimizedMetricLabel ( )
self . _reportMetricLabels = matchPatterns ( self . _reportKeyPatterns , self . _getMetricLabels ( ) )
# Initialize periodic activities ( e . g . , for model result updates )
self . _periodic = self . _initPeriodicActivities ( )
# Create our top - level loop - control iterator
numIters = self . _modelControl . get ( 'iterationCount' , - 1 )
# Are we asked to turn off learning for a certain # of iterations near the
# end ?
learningOffAt = None
iterationCountInferOnly = self . _modelControl . get ( 'iterationCountInferOnly' , 0 )
if iterationCountInferOnly == - 1 :
    self . _model . disableLearning ( )
elif iterationCountInferOnly > 0 :
    assert numIters > iterationCountInferOnly , "when iterationCountInferOnly " "is specified, iterationCount must be greater than " "iterationCountInferOnly."
    learningOffAt = numIters - iterationCountInferOnly
self . __runTaskMainLoop ( numIters , learningOffAt = learningOffAt )
# Perform final operations for model
self . _finalize ( )
return ( self . _cmpReason , None )


signature:
def __runTaskMainLoop ( self , numIters , learningOffAt = None ) :
    """Main loop of the OPF Model Runner .
 Parameters :
 recordIterator : Iterator for counting number of records ( see _ runTask )
 learningOffAt : If not None , learning is turned off when we reach this
 iteration number"""

implementation:
# # Reset sequence states in the model , so it starts looking for a new
# # sequence
self . _model . resetSequenceStates ( )
self . _currentRecordIndex = - 1
while True : # If killed by a terminator , stop running
    if self . _isKilled :
        break
    # If job stops or hypersearch ends , stop running
    if self . _isCanceled :
        break
    # If the process is about to be killed , set as orphaned
    if self . _isInterrupted . isSet ( ) :
        self . __setAsOrphaned ( )
        break
    # If model is mature , stop running ONLY IF we are not the best model
    # for the job . Otherwise , keep running so we can keep returning
    # predictions to the user
    if self . _isMature :
        if not self . _isBestModel :
            self . _cmpReason = self . _jobsDAO . CMPL_REASON_STOPPED
            break
        else :
            self . _cmpReason = self . _jobsDAO . CMPL_REASON_EOF
        # Turn off learning ?
        if learningOffAt is not None and self . _currentRecordIndex == learningOffAt :
            self . _model . disableLearning ( )
    # Read input record . Note that any failure here is a critical JOB failure
    # and results in the job being immediately canceled and marked as
    # failed . The runModelXXX code in hypesearch . utils , if it sees an
    # exception of type utils . JobFailException , will cancel the job and
    # copy the error message into the job record .
    try :
        inputRecord = self . _inputSource . getNextRecordDict ( )
        if self . _currentRecordIndex < 0 :
            self . _inputSource . setTimeout ( 10 )
    except Exception , e :
        raise utils . JobFailException ( ErrorCodes . streamReading , str ( e . args ) , traceback . format_exc ( ) )
    if inputRecord is None : # EOF
        self . _cmpReason = self . _jobsDAO . CMPL_REASON_EOF
        break
    if inputRecord : # Process input record
        self . _currentRecordIndex += 1
        result = self . _model . run ( inputRecord = inputRecord )
        # Compute metrics .
        result . metrics = self . __metricMgr . update ( result )
        # If there are None , use defaults . see MetricsManager . getMetrics ( )
        # TODO remove this when JAVA API server is gone
        if not result . metrics :
            result . metrics = self . __metricMgr . getMetrics ( )
        # Write the result to the output cache . Don ' t write encodings , if they
        # were computed
        if InferenceElement . encodings in result . inferences :
            result . inferences . pop ( InferenceElement . encodings )
        result . sensorInput . dataEncodings = None
        self . _writePrediction ( result )
        # Run periodic activities
        self . _periodic . tick ( )
        if numIters >= 0 and self . _currentRecordIndex >= numIters - 1 :
            break
    else : # Input source returned an empty record .
    # NOTE : This is okay with Stream - based Source ( when it times out
    # waiting for next record ) , but not okay with FileSource , which should
    # always return either with a valid record or None for EOF .
        raise ValueError ( "Got an empty record from FileSource: %r" % inputRecord )


signature:
def _finalize ( self ) :
    """Run final activities after a model has run . These include recording and
 logging the final score"""

implementation:
self . _logger . info ( "Finished: modelID=%r; %r records processed. Performing final activities" , self . _modelID , self . _currentRecordIndex + 1 )
# Dump the experiment metrics at the end of the task
self . _updateModelDBResults ( )
# Check if the current model is the best . Create a milestone if necessary
# If the model has been killed , it is not a candidate for " best model " ,
# and its output cache should be destroyed
if not self . _isKilled :
    self . __updateJobResults ( )
else :
    self . __deleteOutputCache ( self . _modelID )
# Close output stream , if necessary
if self . _predictionLogger :
    self . _predictionLogger . close ( )
# Close input stream , if necessary
if self . _inputSource :
    self . _inputSource . close ( )


signature:
def __createModelCheckpoint ( self ) :
    """Create a checkpoint from the current model , and store it in a dir named
 after checkpoint GUID , and finally store the GUID in the Models DB"""

implementation:
if self . _model is None or self . _modelCheckpointGUID is None :
    return
# Create an output store , if one doesn ' t exist already
if self . _predictionLogger is None :
    self . _createPredictionLogger ( )
predictions = StringIO . StringIO ( )
self . _predictionLogger . checkpoint ( checkpointSink = predictions , maxRows = int ( Configuration . get ( 'nupic.model.checkpoint.maxPredictionRows' ) ) )
self . _model . save ( os . path . join ( self . _experimentDir , str ( self . _modelCheckpointGUID ) ) )
self . _jobsDAO . modelSetFields ( modelID , { 'modelCheckpointId' : str ( self . _modelCheckpointGUID ) } , ignoreUnchanged = True )
self . _logger . info ( "Checkpointed Hypersearch Model: modelID: %r, " "checkpointID: %r" , self . _modelID , checkpointID )
return


signature:
def __deleteModelCheckpoint ( self , modelID ) :
    """Delete the stored checkpoint for the specified modelID . This function is
 called if the current model is now the best model , making the old model ' s
 checkpoint obsolete
 Parameters :
 modelID : The modelID for the checkpoint to delete . This is NOT the
 unique checkpointID"""

implementation:
checkpointID = self . _jobsDAO . modelsGetFields ( modelID , [ 'modelCheckpointId' ] ) [ 0 ]
if checkpointID is None :
    return
try :
    shutil . rmtree ( os . path . join ( self . _experimentDir , str ( self . _modelCheckpointGUID ) ) )
except :
    self . _logger . warn ( "Failed to delete model checkpoint %s. " "Assuming that another worker has already deleted it" , checkpointID )
    return
self . _jobsDAO . modelSetFields ( modelID , { 'modelCheckpointId' : None } , ignoreUnchanged = True )
return


signature:
def _createPredictionLogger ( self ) :
    """Creates the model ' s PredictionLogger object , which is an interface to write
 model results to a permanent storage location"""

implementation:
# Write results to a file
self . _predictionLogger = BasicPredictionLogger ( fields = self . _model . getFieldInfo ( ) , experimentDir = self . _experimentDir , label = "hypersearch-worker" , inferenceType = self . _model . getInferenceType ( ) )
if self . __loggedMetricPatterns :
    metricLabels = self . __metricMgr . getMetricLabels ( )
    loggedMetrics = matchPatterns ( self . __loggedMetricPatterns , metricLabels )
    self . _predictionLogger . setLoggedMetrics ( loggedMetrics )


signature:
def __getOptimizedMetricLabel ( self ) :
    """Get the label for the metric being optimized . This function also caches
 the label in the instance variable self . _ optimizedMetricLabel
 Parameters :
 metricLabels : A sequence of all the labels being computed for this model
 Returns : The label for the metric being optmized over"""

implementation:
matchingKeys = matchPatterns ( [ self . _optimizeKeyPattern ] , self . _getMetricLabels ( ) )
if len ( matchingKeys ) == 0 :
    raise Exception ( "None of the generated metrics match the specified " "optimization pattern: %s. Available metrics are %s" % ( self . _optimizeKeyPattern , self . _getMetricLabels ( ) ) )
elif len ( matchingKeys ) > 1 :
    raise Exception ( "The specified optimization pattern '%s' matches more " "than one metric: %s" % ( self . _optimizeKeyPattern , matchingKeys ) )
return matchingKeys [ 0 ]


signature:
def _getFieldStats ( self ) :
    """Method which returns a dictionary of field statistics received from the
 input source .
 Returns :
 fieldStats : dict of dicts where the first level is the field name and
 the second level is the statistic . ie . fieldStats [ ' pounds ' ] [ ' min ' ]"""

implementation:
fieldStats = dict ( )
fieldNames = self . _inputSource . getFieldNames ( )
for field in fieldNames :
    curStats = dict ( )
    curStats [ 'min' ] = self . _inputSource . getFieldMin ( field )
    curStats [ 'max' ] = self . _inputSource . getFieldMax ( field )
    fieldStats [ field ] = curStats
return fieldStats


signature:
def _updateModelDBResults ( self ) :
    """Retrieves the current results and updates the model ' s record in
 the Model database ."""

implementation:
# Get metrics
metrics = self . _getMetrics ( )
# Extract report metrics that match the requested report REs
reportDict = dict ( [ ( k , metrics [ k ] ) for k in self . _reportMetricLabels ] )
# Extract the report item that matches the optimize key RE
# TODO cache optimizedMetricLabel sooner
metrics = self . _getMetrics ( )
optimizeDict = dict ( )
if self . _optimizeKeyPattern is not None :
    optimizeDict [ self . _optimizedMetricLabel ] = metrics [ self . _optimizedMetricLabel ]
# Update model results
results = json . dumps ( ( metrics , optimizeDict ) )
self . _jobsDAO . modelUpdateResults ( self . _modelID , results = results , metricValue = optimizeDict . values ( ) [ 0 ] , numRecords = ( self . _currentRecordIndex + 1 ) )
self . _logger . debug ( "Model Results: modelID=%s; numRecords=%s; results=%s" % ( self . _modelID , self . _currentRecordIndex + 1 , results ) )
return


signature:
def __updateJobResultsPeriodic ( self ) :
    """Periodic check to see if this is the best model . This should only have an
 effect if this is the * first * model to report its progress"""

implementation:
if self . _isBestModelStored and not self . _isBestModel :
    return
while True :
    jobResultsStr = self . _jobsDAO . jobGetFields ( self . _jobID , [ 'results' ] ) [ 0 ]
    if jobResultsStr is None :
        jobResults = { }
    else :
        self . _isBestModelStored = True
        if not self . _isBestModel :
            return
        jobResults = json . loads ( jobResultsStr )
    bestModel = jobResults . get ( 'bestModel' , None )
    bestMetric = jobResults . get ( 'bestValue' , None )
    isSaved = jobResults . get ( 'saved' , False )
    # If there is a best model , and it is not the same as the current model
    # we should wait till we have processed all of our records to see if
    # we are the the best
    if ( bestModel is not None ) and ( self . _modelID != bestModel ) :
        self . _isBestModel = False
        return
    # Make sure prediction output stream is ready before we present our model
    # as " bestModel " ; sometimes this takes a long time , so update the model ' s
    # timestamp to help avoid getting orphaned
    self . __flushPredictionCache ( )
    self . _jobsDAO . modelUpdateTimestamp ( self . _modelID )
    metrics = self . _getMetrics ( )
    jobResults [ 'bestModel' ] = self . _modelID
    jobResults [ 'bestValue' ] = metrics [ self . _optimizedMetricLabel ]
    jobResults [ 'metrics' ] = metrics
    jobResults [ 'saved' ] = False
    newResults = json . dumps ( jobResults )
    isUpdated = self . _jobsDAO . jobSetFieldIfEqual ( self . _jobID , fieldName = 'results' , curValue = jobResultsStr , newValue = newResults )
    if isUpdated or ( not isUpdated and newResults == jobResultsStr ) :
        self . _isBestModel = True
        break


signature:
def __checkIfBestCompletedModel ( self ) :
    """Reads the current " best model " for the job and returns whether or not the
 current model is better than the " best model " stored for the job
 Returns : ( isBetter , storedBest , origResultsStr )
 isBetter :
 True if the current model is better than the stored " best model "
 storedResults :
 A dict of the currently stored results in the jobs table record
 origResultsStr :
 The json - encoded string that currently resides in the " results " field
 of the jobs record ( used to create atomicity )"""

implementation:
jobResultsStr = self . _jobsDAO . jobGetFields ( self . _jobID , [ 'results' ] ) [ 0 ]
if jobResultsStr is None :
    jobResults = { }
else :
    jobResults = json . loads ( jobResultsStr )
isSaved = jobResults . get ( 'saved' , False )
bestMetric = jobResults . get ( 'bestValue' , None )
currentMetric = self . _getMetrics ( ) [ self . _optimizedMetricLabel ]
self . _isBestModel = ( not isSaved ) or ( currentMetric < bestMetric )
return self . _isBestModel , jobResults , jobResultsStr


signature:
def __updateJobResults ( self ) :
    """Check if this is the best model
 If so :
 1 ) Write it ' s checkpoint
 2 ) Record this model as the best
 3 ) Delete the previous best ' s output cache
 Otherwise :
 1 ) Delete our output cache"""

implementation:
isSaved = False
while True :
    self . _isBestModel , jobResults , jobResultsStr = self . __checkIfBestCompletedModel ( )
    # If the current model is the best :
    # 1 ) Save the model ' s predictions
    # 2 ) Checkpoint the model state
    # 3 ) Update the results for the job
    if self . _isBestModel : # Save the current model and its results
        if not isSaved :
            self . __flushPredictionCache ( )
            self . _jobsDAO . modelUpdateTimestamp ( self . _modelID )
            self . __createModelCheckpoint ( )
            self . _jobsDAO . modelUpdateTimestamp ( self . _modelID )
            isSaved = True
        # Now record the model as the best for the job
        prevBest = jobResults . get ( 'bestModel' , None )
        prevWasSaved = jobResults . get ( 'saved' , False )
        # If the current model is the best , it shouldn ' t already be checkpointed
        if prevBest == self . _modelID :
            assert not prevWasSaved
        metrics = self . _getMetrics ( )
        jobResults [ 'bestModel' ] = self . _modelID
        jobResults [ 'bestValue' ] = metrics [ self . _optimizedMetricLabel ]
        jobResults [ 'metrics' ] = metrics
        jobResults [ 'saved' ] = True
        isUpdated = self . _jobsDAO . jobSetFieldIfEqual ( self . _jobID , fieldName = 'results' , curValue = jobResultsStr , newValue = json . dumps ( jobResults ) )
        if isUpdated :
            if prevWasSaved :
                self . __deleteOutputCache ( prevBest )
                self . _jobsDAO . modelUpdateTimestamp ( self . _modelID )
                self . __deleteModelCheckpoint ( prevBest )
                self . _jobsDAO . modelUpdateTimestamp ( self . _modelID )
            self . _logger . info ( "Model %d chosen as best model" , self . _modelID )
            break
    # If the current model is not the best , delete its outputs
    else : # NOTE : we update model timestamp around these occasionally - lengthy
    # operations to help prevent the model from becoming orphaned
        self . __deleteOutputCache ( self . _modelID )
        self . _jobsDAO . modelUpdateTimestamp ( self . _modelID )
        self . __deleteModelCheckpoint ( self . _modelID )
        self . _jobsDAO . modelUpdateTimestamp ( self . _modelID )
        break


signature:
def _writePrediction ( self , result ) :
    """Writes the results of one iteration of a model . The results are written to
 this ModelRunner ' s in - memory cache unless this model is the " best model " for
 the job . If this model is the " best model " , the predictions are written out
 to a permanent store via a prediction output stream instance
 Parameters :
 result : A opf _ utils . ModelResult object , which contains the input and
 output for this iteration"""

implementation:
self . __predictionCache . append ( result )
if self . _isBestModel :
    self . __flushPredictionCache ( )


signature:
def __flushPredictionCache ( self ) :
    """Writes the contents of this model ' s in - memory prediction cache to a permanent
 store via the prediction output stream instance"""

implementation:
if not self . __predictionCache :
    return
# Create an output store , if one doesn ' t exist already
if self . _predictionLogger is None :
    self . _createPredictionLogger ( )
startTime = time . time ( )
self . _predictionLogger . writeRecords ( self . __predictionCache , progressCB = self . __writeRecordsCallback )
self . _logger . info ( "Flushed prediction cache; numrows=%s; elapsed=%s sec." , len ( self . __predictionCache ) , time . time ( ) - startTime )
self . __predictionCache . clear ( )


signature:
def __deleteOutputCache ( self , modelID ) :
    """Delete ' s the output cache associated with the given modelID . This actually
 clears up the resources associated with the cache , rather than deleting al
 the records in the cache
 Parameters :
 modelID : The id of the model whose output cache is being deleted"""

implementation:
# If this is our output , we should close the connection
if modelID == self . _modelID and self . _predictionLogger is not None :
    self . _predictionLogger . close ( )
    del self . __predictionCache
    self . _predictionLogger = None
    self . __predictionCache = None


signature:
def _initPeriodicActivities ( self ) :
    """Creates and returns a PeriodicActivityMgr instance initialized with
 our periodic activities
 Parameters :
 retval : a PeriodicActivityMgr instance"""

implementation:
# Activity to update the metrics for this model
# in the models table
updateModelDBResults = PeriodicActivityRequest ( repeating = True , period = 100 , cb = self . _updateModelDBResults )
updateJobResults = PeriodicActivityRequest ( repeating = True , period = 100 , cb = self . __updateJobResultsPeriodic )
checkCancelation = PeriodicActivityRequest ( repeating = True , period = 50 , cb = self . __checkCancelation )
checkMaturity = PeriodicActivityRequest ( repeating = True , period = 10 , cb = self . __checkMaturity )
# Do an initial update of the job record after 2 iterations to make
# sure that it is populated with something without having to wait too long
updateJobResultsFirst = PeriodicActivityRequest ( repeating = False , period = 2 , cb = self . __updateJobResultsPeriodic )
periodicActivities = [ updateModelDBResults , updateJobResultsFirst , updateJobResults , checkCancelation ]
if self . _isMaturityEnabled :
    periodicActivities . append ( checkMaturity )
return PeriodicActivityMgr ( requestedActivities = periodicActivities )


signature:
def __checkCancelation ( self ) :
    """Check if the cancelation flag has been set for this model
 in the Model DB"""

implementation:
# Update a hadoop job counter at least once every 600 seconds so it doesn ' t
# think our map task is dead
print >> sys . stderr , "reporter:counter:HypersearchWorker,numRecords,50"
# See if the job got cancelled
jobCancel = self . _jobsDAO . jobGetFields ( self . _jobID , [ 'cancel' ] ) [ 0 ]
if jobCancel :
    self . _cmpReason = ClientJobsDAO . CMPL_REASON_KILLED
    self . _isCanceled = True
    self . _logger . info ( "Model %s canceled because Job %s was stopped." , self . _modelID , self . _jobID )
else :
    stopReason = self . _jobsDAO . modelsGetFields ( self . _modelID , [ 'engStop' ] ) [ 0 ]
    if stopReason is None :
        pass
    elif stopReason == ClientJobsDAO . STOP_REASON_KILLED :
        self . _cmpReason = ClientJobsDAO . CMPL_REASON_KILLED
        self . _isKilled = True
        self . _logger . info ( "Model %s canceled because it was killed by hypersearch" , self . _modelID )
    elif stopReason == ClientJobsDAO . STOP_REASON_STOPPED :
        self . _cmpReason = ClientJobsDAO . CMPL_REASON_STOPPED
        self . _isCanceled = True
        self . _logger . info ( "Model %s stopped because hypersearch ended" , self . _modelID )
    else :
        raise RuntimeError ( "Unexpected stop reason encountered: %s" % ( stopReason ) )


signature:
def __checkMaturity ( self ) :
    """Save the current metric value and see if the model ' s performance has
 ' leveled off . ' We do this by looking at some number of previous number of
 recordings"""

implementation:
if self . _currentRecordIndex + 1 < self . _MIN_RECORDS_TO_BE_BEST :
    return
# If we are already mature , don ' t need to check anything
if self . _isMature :
    return
metric = self . _getMetrics ( ) [ self . _optimizedMetricLabel ]
self . _metricRegression . addPoint ( x = self . _currentRecordIndex , y = metric )
# Perform a linear regression to see if the error is leveled off
# pctChange = self . _ metricRegression . getPctChange ( )
# if pctChange is not None and abs ( pctChange ) < = self . _ MATURITY _ MAX _ CHANGE :
pctChange , absPctChange = self . _metricRegression . getPctChanges ( )
if pctChange is not None and absPctChange <= self . _MATURITY_MAX_CHANGE :
    self . _jobsDAO . modelSetFields ( self . _modelID , { 'engMatured' : True } )
    # TODO : Don ' t stop if we are currently the best model . Also , if we
    # are still running after maturity , we have to periodically check to
    # see if we are still the best model . As soon we lose to some other
    # model , then we should stop at that point .
    self . _cmpReason = ClientJobsDAO . CMPL_REASON_STOPPED
    self . _isMature = True
    self . _logger . info ( "Model %d has matured (pctChange=%s, n=%d). \n" "Scores = %s\n" "Stopping execution" , self . _modelID , pctChange , self . _MATURITY_NUM_POINTS , self . _metricRegression . _window )


signature:
def __setAsOrphaned ( self ) :
    """Sets the current model as orphaned . This is called when the scheduler is
 about to kill the process to reallocate the worker to a different process ."""

implementation:
cmplReason = ClientJobsDAO . CMPL_REASON_ORPHAN
cmplMessage = "Killed by Scheduler"
self . _jobsDAO . modelSetCompleted ( self . _modelID , cmplReason , cmplMessage )


signature:
def readStateFromDB ( self ) :
    """Set our state to that obtained from the engWorkerState field of the
 job record .
 Parameters :
 stateJSON : JSON encoded state from job record"""

implementation:
self . _priorStateJSON = self . _hsObj . _cjDAO . jobGetFields ( self . _hsObj . _jobID , [ 'engWorkerState' ] ) [ 0 ]
# Init if no prior state yet
if self . _priorStateJSON is None :
    swarms = dict ( )
    # Fast Swarm , first and only sprint has one swarm for each field
    # in fixedFields
    if self . _hsObj . _fixedFields is not None :
        print self . _hsObj . _fixedFields
        encoderSet = [ ]
        for field in self . _hsObj . _fixedFields :
            if field == '_classifierInput' :
                continue
            encoderName = self . getEncoderKeyFromName ( field )
            assert encoderName in self . _hsObj . _encoderNames , "The field '%s' " " specified in the fixedFields list is not present in this " " model." % ( field )
            encoderSet . append ( encoderName )
        encoderSet . sort ( )
        swarms [ '.' . join ( encoderSet ) ] = { 'status' : 'active' , 'bestModelId' : None , 'bestErrScore' : None , 'sprintIdx' : 0 , }
    # Temporal prediction search , first sprint has N swarms of 1 field each ,
    # the predicted field may or may not be that one field .
    elif self . _hsObj . _searchType == HsSearchType . temporal :
        for encoderName in self . _hsObj . _encoderNames :
            swarms [ encoderName ] = { 'status' : 'active' , 'bestModelId' : None , 'bestErrScore' : None , 'sprintIdx' : 0 , }
    # Classification prediction search , first sprint has N swarms of 1 field
    # each where this field can NOT be the predicted field .
    elif self . _hsObj . _searchType == HsSearchType . classification :
        for encoderName in self . _hsObj . _encoderNames :
            if encoderName == self . _hsObj . _predictedFieldEncoder :
                continue
            swarms [ encoderName ] = { 'status' : 'active' , 'bestModelId' : None , 'bestErrScore' : None , 'sprintIdx' : 0 , }
    # Legacy temporal . This is either a model that uses reconstruction or
    # an older multi - step model that doesn ' t have a separate
    # ' classifierOnly ' encoder for the predicted field . Here , the predicted
    # field must ALWAYS be present and the first sprint tries the predicted
    # field only
    elif self . _hsObj . _searchType == HsSearchType . legacyTemporal :
        swarms [ self . _hsObj . _predictedFieldEncoder ] = { 'status' : 'active' , 'bestModelId' : None , 'bestErrScore' : None , 'sprintIdx' : 0 , }
    else :
        raise RuntimeError ( "Unsupported search type: %s" % ( self . _hsObj . _searchType ) )
    # Initialize the state .
    self . _state = dict ( # The last time the state was updated by a worker .
    lastUpdateTime = time . time ( ) , # Set from within setSwarmState ( ) if we detect that the sprint we just
    # completed did worse than a prior sprint . This stores the index of
    # the last good sprint .
    lastGoodSprint = None , # Set from within setSwarmState ( ) if lastGoodSprint is True and all
    # sprints have completed .
    searchOver = False , # This is a summary of the active swarms - this information can also
    # be obtained from the swarms entry that follows , but is summarized here
    # for easier reference when viewing the state as presented by
    # log messages and prints of the hsState data structure ( by
    # permutations _ runner ) .
    activeSwarms = swarms . keys ( ) , # All the swarms that have been created so far .
    swarms = swarms , # All the sprints that have completed or are in progress .
    sprints = [ { 'status' : 'active' , 'bestModelId' : None , 'bestErrScore' : None } ] , # The list of encoders we have " blacklisted " because they
    # performed so poorly .
    blackListedEncoders = [ ] , )
    # This will do nothing if the value of engWorkerState is not still None .
    self . _hsObj . _cjDAO . jobSetFieldIfEqual ( self . _hsObj . _jobID , 'engWorkerState' , json . dumps ( self . _state ) , None )
    self . _priorStateJSON = self . _hsObj . _cjDAO . jobGetFields ( self . _hsObj . _jobID , [ 'engWorkerState' ] ) [ 0 ]
    assert ( self . _priorStateJSON is not None )
# Read state from the database
self . _state = json . loads ( self . _priorStateJSON )
self . _dirty = False


signature:
def writeStateToDB ( self ) :
    """Update the state in the job record with our local changes ( if any ) .
 If we don ' t have the latest state in our priorStateJSON , then re - load
 in the latest state and return False . If we were successful writing out
 our changes , return True
 Parameters :
 retval : True if we were successful writing out our changes
 False if our priorState is not the latest that was in the DB .
 In this case , we will re - load our state from the DB"""

implementation:
# If no changes , do nothing
if not self . _dirty :
    return True
# Set the update time
self . _state [ 'lastUpdateTime' ] = time . time ( )
newStateJSON = json . dumps ( self . _state )
success = self . _hsObj . _cjDAO . jobSetFieldIfEqual ( self . _hsObj . _jobID , 'engWorkerState' , str ( newStateJSON ) , str ( self . _priorStateJSON ) )
if success :
    self . logger . debug ( "Success changing hsState to: \n%s " % ( pprint . pformat ( self . _state , indent = 4 ) ) )
    self . _priorStateJSON = newStateJSON
# If no success , read in the current state from the DB
else :
    self . logger . debug ( "Failed to change hsState to: \n%s " % ( pprint . pformat ( self . _state , indent = 4 ) ) )
    self . _priorStateJSON = self . _hsObj . _cjDAO . jobGetFields ( self . _hsObj . _jobID , [ 'engWorkerState' ] ) [ 0 ]
    self . _state = json . loads ( self . _priorStateJSON )
    self . logger . info ( "New hsState has been set by some other worker to: " " \n%s" % ( pprint . pformat ( self . _state , indent = 4 ) ) )
return success


signature:
def getFieldContributions ( self ) :
    """Return the field contributions statistics .
 Parameters :
 retval : Dictionary where the keys are the field names and the values
 are how much each field contributed to the best score ."""

implementation:
# in the fast swarm , there is only 1 sprint and field contributions are
# not defined
if self . _hsObj . _fixedFields is not None :
    return dict ( ) , dict ( )
# Get the predicted field encoder name
predictedEncoderName = self . _hsObj . _predictedFieldEncoder
# Collect all the single field scores
fieldScores = [ ]
for swarmId , info in self . _state [ 'swarms' ] . iteritems ( ) :
    encodersUsed = swarmId . split ( '.' )
    if len ( encodersUsed ) != 1 :
        continue
    field = self . getEncoderNameFromKey ( encodersUsed [ 0 ] )
    bestScore = info [ 'bestErrScore' ]
    # If the bestScore is None , this swarm hasn ' t completed yet ( this could
    # happen if we ' re exiting because of maxModels ) , so look up the best
    # score so far
    if bestScore is None :
        ( _modelId , bestScore ) = self . _hsObj . _resultsDB . bestModelIdAndErrScore ( swarmId )
    fieldScores . append ( ( bestScore , field ) )
# If we only have 1 field that was tried in the first sprint , then use that
# as the base and get the contributions from the fields in the next sprint .
if self . _hsObj . _searchType == HsSearchType . legacyTemporal :
    assert ( len ( fieldScores ) == 1 )
    ( baseErrScore , baseField ) = fieldScores [ 0 ]
    for swarmId , info in self . _state [ 'swarms' ] . iteritems ( ) :
        encodersUsed = swarmId . split ( '.' )
        if len ( encodersUsed ) != 2 :
            continue
        fields = [ self . getEncoderNameFromKey ( name ) for name in encodersUsed ]
        fields . remove ( baseField )
        fieldScores . append ( ( info [ 'bestErrScore' ] , fields [ 0 ] ) )
# The first sprint tried a bunch of fields , pick the worst performing one
# ( within the top self . _ hsObj . _ maxBranching ones ) as the base
else :
    fieldScores . sort ( reverse = True )
    # If maxBranching was specified , pick the worst performing field within
    # the top maxBranching + 1 fields as our base , which will give that field
    # a contribution of 0.
    if self . _hsObj . _maxBranching > 0 and len ( fieldScores ) > self . _hsObj . _maxBranching :
        baseErrScore = fieldScores [ - self . _hsObj . _maxBranching - 1 ] [ 0 ]
    else :
        baseErrScore = fieldScores [ 0 ] [ 0 ]
# Prepare and return the fieldContributions dict
pctFieldContributionsDict = dict ( )
absFieldContributionsDict = dict ( )
# If we have no base score , can ' t compute field contributions . This can
# happen when we exit early due to maxModels or being cancelled
if baseErrScore is not None : # If the base error score is 0 , we can ' t compute a percent difference
# off of it , so move it to a very small float
    if abs ( baseErrScore ) < 0.00001 :
        baseErrScore = 0.00001
    for ( errScore , field ) in fieldScores :
        if errScore is not None :
            pctBetter = ( baseErrScore - errScore ) * 100.0 / baseErrScore
        else :
            pctBetter = 0.0
            errScore = baseErrScore
        # for absFieldContribution
        pctFieldContributionsDict [ field ] = pctBetter
        absFieldContributionsDict [ field ] = baseErrScore - errScore
self . logger . debug ( "FieldContributions: %s" % ( pctFieldContributionsDict ) )
return pctFieldContributionsDict , absFieldContributionsDict


signature:
def getAllSwarms ( self , sprintIdx ) :
    """Return the list of all swarms in the given sprint .
 Parameters :
 retval : list of active swarm Ids in the given sprint"""

implementation:
swarmIds = [ ]
for swarmId , info in self . _state [ 'swarms' ] . iteritems ( ) :
    if info [ 'sprintIdx' ] == sprintIdx :
        swarmIds . append ( swarmId )
return swarmIds


signature:
def getCompletedSwarms ( self ) :
    """Return the list of all completed swarms .
 Parameters :
 retval : list of active swarm Ids"""

implementation:
swarmIds = [ ]
for swarmId , info in self . _state [ 'swarms' ] . iteritems ( ) :
    if info [ 'status' ] == 'completed' :
        swarmIds . append ( swarmId )
return swarmIds


signature:
def getCompletingSwarms ( self ) :
    """Return the list of all completing swarms .
 Parameters :
 retval : list of active swarm Ids"""

implementation:
swarmIds = [ ]
for swarmId , info in self . _state [ 'swarms' ] . iteritems ( ) :
    if info [ 'status' ] == 'completing' :
        swarmIds . append ( swarmId )
return swarmIds


signature:
def bestModelInSprint ( self , sprintIdx ) :
    """Return the best model ID and it ' s errScore from the given sprint ,
 which may still be in progress . This returns the best score from all models
 in the sprint which have matured so far .
 Parameters :
 retval : ( modelId , errScore )"""

implementation:
# Get all the swarms in this sprint
swarms = self . getAllSwarms ( sprintIdx )
# Get the best model and score from each swarm
bestModelId = None
bestErrScore = numpy . inf
for swarmId in swarms :
    ( modelId , errScore ) = self . _hsObj . _resultsDB . bestModelIdAndErrScore ( swarmId )
    if errScore < bestErrScore :
        bestModelId = modelId
        bestErrScore = errScore
return ( bestModelId , bestErrScore )


signature:
def setSwarmState ( self , swarmId , newStatus ) :
    """Change the given swarm ' s state to ' newState ' . If ' newState ' is
 ' completed ' , then bestModelId and bestErrScore must be provided .
 Parameters :
 swarmId : swarm Id
 newStatus : new status , either ' active ' , ' completing ' , ' completed ' , or
 ' killed '"""

implementation:
assert ( newStatus in [ 'active' , 'completing' , 'completed' , 'killed' ] )
# Set the swarm status
swarmInfo = self . _state [ 'swarms' ] [ swarmId ]
if swarmInfo [ 'status' ] == newStatus :
    return
# If some other worker noticed it as completed , setting it to completing
# is obviously old information . . . .
if swarmInfo [ 'status' ] == 'completed' and newStatus == 'completing' :
    return
self . _dirty = True
swarmInfo [ 'status' ] = newStatus
if newStatus == 'completed' :
    ( modelId , errScore ) = self . _hsObj . _resultsDB . bestModelIdAndErrScore ( swarmId )
    swarmInfo [ 'bestModelId' ] = modelId
    swarmInfo [ 'bestErrScore' ] = errScore
# If no longer active , remove it from the activeSwarms entry
if newStatus != 'active' and swarmId in self . _state [ 'activeSwarms' ] :
    self . _state [ 'activeSwarms' ] . remove ( swarmId )
# If new status is ' killed ' , kill off any running particles in that swarm
if newStatus == 'killed' :
    self . _hsObj . killSwarmParticles ( swarmId )
# In case speculative particles are enabled , make sure we generate a new
# swarm at this time if all of the swarms in the current sprint have
# completed . This will insure that we don ' t mark the sprint as completed
# before we ' ve created all the possible swarms .
sprintIdx = swarmInfo [ 'sprintIdx' ]
self . isSprintActive ( sprintIdx )
# Update the sprint status . Check all the swarms that belong to this sprint .
# If they are all completed , the sprint is completed .
sprintInfo = self . _state [ 'sprints' ] [ sprintIdx ]
statusCounts = dict ( active = 0 , completing = 0 , completed = 0 , killed = 0 )
bestModelIds = [ ]
bestErrScores = [ ]
for info in self . _state [ 'swarms' ] . itervalues ( ) :
    if info [ 'sprintIdx' ] != sprintIdx :
        continue
    statusCounts [ info [ 'status' ] ] += 1
    if info [ 'status' ] == 'completed' :
        bestModelIds . append ( info [ 'bestModelId' ] )
        bestErrScores . append ( info [ 'bestErrScore' ] )
if statusCounts [ 'active' ] > 0 :
    sprintStatus = 'active'
elif statusCounts [ 'completing' ] > 0 :
    sprintStatus = 'completing'
else :
    sprintStatus = 'completed'
sprintInfo [ 'status' ] = sprintStatus
# If the sprint is complete , get the best model from all of its swarms and
# store that as the sprint best
if sprintStatus == 'completed' :
    if len ( bestErrScores ) > 0 :
        whichIdx = numpy . array ( bestErrScores ) . argmin ( )
        sprintInfo [ 'bestModelId' ] = bestModelIds [ whichIdx ]
        sprintInfo [ 'bestErrScore' ] = bestErrScores [ whichIdx ]
    else : # This sprint was empty , most likely because all particles were
    # killed . Give it a huge error score
        sprintInfo [ 'bestModelId' ] = 0
        sprintInfo [ 'bestErrScore' ] = numpy . inf
    # See if our best err score got NO BETTER as compared to a previous
    # sprint . If so , stop exploring subsequent sprints ( lastGoodSprint
    # is no longer None ) .
    bestPrior = numpy . inf
    for idx in range ( sprintIdx ) :
        if self . _state [ 'sprints' ] [ idx ] [ 'status' ] == 'completed' :
            ( _ , errScore ) = self . bestModelInCompletedSprint ( idx )
            if errScore is None :
                errScore = numpy . inf
        else :
            errScore = numpy . inf
        if errScore < bestPrior :
            bestPrior = errScore
    if sprintInfo [ 'bestErrScore' ] >= bestPrior :
        self . _state [ 'lastGoodSprint' ] = sprintIdx - 1
    # If ALL sprints up to the last good one are done , the search is now over
    if self . _state [ 'lastGoodSprint' ] is not None and not self . anyGoodSprintsActive ( ) :
        self . _state [ 'searchOver' ] = True


signature:
def anyGoodSprintsActive ( self ) :
    """Return True if there are any more good sprints still being explored .
 A ' good ' sprint is one that is earlier than where we detected an increase
 in error from sprint to subsequent sprint ."""

implementation:
if self . _state [ 'lastGoodSprint' ] is not None :
    goodSprints = self . _state [ 'sprints' ] [ 0 : self . _state [ 'lastGoodSprint' ] + 1 ]
else :
    goodSprints = self . _state [ 'sprints' ]
for sprint in goodSprints :
    if sprint [ 'status' ] == 'active' :
        anyActiveSprints = True
        break
else :
    anyActiveSprints = False
return anyActiveSprints


signature:
def isSprintCompleted ( self , sprintIdx ) :
    """Return True if the given sprint has completed ."""

implementation:
numExistingSprints = len ( self . _state [ 'sprints' ] )
if sprintIdx >= numExistingSprints :
    return False
return ( self . _state [ 'sprints' ] [ sprintIdx ] [ 'status' ] == 'completed' )


signature:
def killUselessSwarms ( self ) :
    """See if we can kill off some speculative swarms . If an earlier sprint
 has finally completed , we can now tell which fields should * really * be present
 in the sprints we ' ve already started due to speculation , and kill off the
 swarms that should not have been included ."""

implementation:
# Get number of existing sprints
numExistingSprints = len ( self . _state [ 'sprints' ] )
# Should we bother killing useless swarms ?
if self . _hsObj . _searchType == HsSearchType . legacyTemporal :
    if numExistingSprints <= 2 :
        return
else :
    if numExistingSprints <= 1 :
        return
# Form completedSwarms as a list of tuples , each tuple contains :
# ( swarmName , swarmState , swarmBestErrScore )
# ex . completedSwarms :
# [ ( ' a ' , { . . . } , 1.4 ) ,
# ( ' b ' , { . . . } , 2.0 ) ,
# ( ' c ' , { . . . } , 3.0 ) ]
completedSwarms = self . getCompletedSwarms ( )
completedSwarms = [ ( swarm , self . _state [ "swarms" ] [ swarm ] , self . _state [ "swarms" ] [ swarm ] [ "bestErrScore" ] ) for swarm in completedSwarms ]
# Form the completedMatrix . Each row corresponds to a sprint . Each row
# contains the list of swarm tuples that belong to that sprint , sorted
# by best score . Each swarm tuple contains ( swarmName , swarmState ,
# swarmBestErrScore ) .
# ex . completedMatrix :
# [ ( ( ' a ' , { . . . } , 1.4 ) , ( ' b ' , { . . . } , 2.0 ) , ( ' c ' , { . . . } , 3.0 ) ) ,
# ( ( ' a . b ' , { . . . } , 3.0 ) , ( ' b . c ' , { . . . } , 4.0 ) ) ]
completedMatrix = [ [ ] for i in range ( numExistingSprints ) ]
for swarm in completedSwarms :
    completedMatrix [ swarm [ 1 ] [ "sprintIdx" ] ] . append ( swarm )
for sprint in completedMatrix :
    sprint . sort ( key = itemgetter ( 2 ) )
# Form activeSwarms as a list of tuples , each tuple contains :
# ( swarmName , swarmState , swarmBestErrScore )
# Include all activeSwarms and completingSwarms
# ex . activeSwarms :
# [ ( ' d ' , { . . . } , 1.4 ) ,
# ( ' e ' , { . . . } , 2.0 ) ,
# ( ' f ' , { . . . } , 3.0 ) ]
activeSwarms = self . getActiveSwarms ( )
# Append the completing swarms
activeSwarms . extend ( self . getCompletingSwarms ( ) )
activeSwarms = [ ( swarm , self . _state [ "swarms" ] [ swarm ] , self . _state [ "swarms" ] [ swarm ] [ "bestErrScore" ] ) for swarm in activeSwarms ]
# Form the activeMatrix . Each row corresponds to a sprint . Each row
# contains the list of swarm tuples that belong to that sprint , sorted
# by best score . Each swarm tuple contains ( swarmName , swarmState ,
# swarmBestErrScore )
# ex . activeMatrix :
# [ ( ( ' d ' , { . . . } , 1.4 ) , ( ' e ' , { . . . } , 2.0 ) , ( ' f ' , { . . . } , 3.0 ) ) ,
# ( ( ' d . e ' , { . . . } , 3.0 ) , ( ' e . f ' , { . . . } , 4.0 ) ) ]
activeMatrix = [ [ ] for i in range ( numExistingSprints ) ]
for swarm in activeSwarms :
    activeMatrix [ swarm [ 1 ] [ "sprintIdx" ] ] . append ( swarm )
for sprint in activeMatrix :
    sprint . sort ( key = itemgetter ( 2 ) )
# Figure out which active swarms to kill
toKill = [ ]
for i in range ( 1 , numExistingSprints ) :
    for swarm in activeMatrix [ i ] :
        curSwarmEncoders = swarm [ 0 ] . split ( "." )
        # If previous sprint is complete , get the best swarm and kill all active
        # sprints that are not supersets
        if ( len ( activeMatrix [ i - 1 ] ) == 0 ) : # If we are trying all possible 3 field combinations , don ' t kill any
        # off in sprint 2
            if i == 2 and ( self . _hsObj . _tryAll3FieldCombinations or self . _hsObj . _tryAll3FieldCombinationsWTimestamps ) :
                pass
            else :
                bestInPrevious = completedMatrix [ i - 1 ] [ 0 ]
                bestEncoders = bestInPrevious [ 0 ] . split ( '.' )
                for encoder in bestEncoders :
                    if not encoder in curSwarmEncoders :
                        toKill . append ( swarm )
        # if there are more than two completed encoders sets that are complete and
        # are worse than at least one active swarm in the previous sprint . Remove
        # any combinations that have any pair of them since they cannot have the best encoder .
        # elif ( len ( completedMatrix [ i - 1 ] ) > 1 ) :
        # for completedSwarm in completedMatrix [ i - 1 ] :
        # activeMatrix [ i - 1 ] [ 0 ] [ 2 ] < completed
# Mark the bad swarms as killed
if len ( toKill ) > 0 :
    print "ParseMe: Killing encoders:" + str ( toKill )
for swarm in toKill :
    self . setSwarmState ( swarm [ 0 ] , "killed" )
return


signature:
def isSprintActive ( self , sprintIdx ) :
    """If the given sprint exists and is active , return active = True .
 If the sprint does not exist yet , this call will create it ( and return
 active = True ) . If it already exists , but is completing or complete , return
 active = False .
 If sprintIdx is past the end of the possible sprints , return
 active = False , noMoreSprints = True
 IMPORTANT : When speculative particles are enabled , this call has some
 special processing to handle speculative sprints :
 * When creating a new speculative sprint ( creating sprint N before
 sprint N - 1 has completed ) , it initially only puts in only ONE swarm into
 the sprint .
 * Every time it is asked if sprint N is active , it also checks to see if
 it is time to add another swarm to the sprint , and adds a new swarm if
 appropriate before returning active = True
 * We decide it is time to add a new swarm to a speculative sprint when ALL
 of the currently active swarms in the sprint have all the workers they
 need ( number of running ( not mature ) particles is _ minParticlesPerSwarm ) .
 This means that we have capacity to run additional particles in a new
 swarm .
 It is expected that the sprints will be checked IN ORDER from 0 on up . ( It
 is an error not to ) The caller should always try to allocate from the first
 active sprint it finds . If it can ' t , then it can call this again to
 find / create the next active sprint .
 Parameters :
 retval : ( active , noMoreSprints )
 active : True if the given sprint is active
 noMoreSprints : True if there are no more sprints possible"""

implementation:
while True :
    numExistingSprints = len ( self . _state [ 'sprints' ] )
    # If this sprint already exists , see if it is active
    if sprintIdx <= numExistingSprints - 1 : # With speculation off , it ' s simple , just return whether or not the
    # asked for sprint has active status
        if not self . _hsObj . _speculativeParticles :
            active = ( self . _state [ 'sprints' ] [ sprintIdx ] [ 'status' ] == 'active' )
            return ( active , False )
        # With speculation on , if the sprint is still marked active , we also
        # need to see if it ' s time to add a new swarm to it .
        else :
            active = ( self . _state [ 'sprints' ] [ sprintIdx ] [ 'status' ] == 'active' )
            if not active :
                return ( active , False )
            # See if all of the existing swarms are at capacity ( have all the
            # workers they need ) :
            activeSwarmIds = self . getActiveSwarms ( sprintIdx )
            swarmSizes = [ self . _hsObj . _resultsDB . getParticleInfos ( swarmId , matured = False ) [ 0 ] for swarmId in activeSwarmIds ]
            notFullSwarms = [ len ( swarm ) for swarm in swarmSizes if len ( swarm ) < self . _hsObj . _minParticlesPerSwarm ]
            # If some swarms have room return that the swarm is active .
            if len ( notFullSwarms ) > 0 :
                return ( True , False )
            # If the existing swarms are at capacity , we will fall through to the
            # logic below which tries to add a new swarm to the sprint .
    # Stop creating new sprints ?
    if self . _state [ 'lastGoodSprint' ] is not None :
        return ( False , True )
    # if fixedFields is set , we are running a fast swarm and only run sprint0
    if self . _hsObj . _fixedFields is not None :
        return ( False , True )
    # Get the best model ( if there is one ) from the prior sprint . That gives
    # us the base encoder set for the next sprint . For sprint zero make sure
    # it does not take the last sprintidx because of wrapping .
    if sprintIdx > 0 and self . _state [ 'sprints' ] [ sprintIdx - 1 ] [ 'status' ] == 'completed' :
        ( bestModelId , _ ) = self . bestModelInCompletedSprint ( sprintIdx - 1 )
        ( particleState , _ , _ , _ , _ ) = self . _hsObj . _resultsDB . getParticleInfo ( bestModelId )
        bestSwarmId = particleState [ 'swarmId' ]
        baseEncoderSets = [ bestSwarmId . split ( '.' ) ]
    # If there is no best model yet , then use all encoder sets from the prior
    # sprint that were not killed
    else :
        bestSwarmId = None
        particleState = None
        # Build up more combinations , using ALL of the sets in the current
        # sprint .
        baseEncoderSets = [ ]
        for swarmId in self . getNonKilledSwarms ( sprintIdx - 1 ) :
            baseEncoderSets . append ( swarmId . split ( '.' ) )
    # Which encoders should we add to the current base set ?
    encoderAddSet = [ ]
    # If we have constraints on how many fields we carry forward into
    # subsequent sprints ( either nupic . hypersearch . max . field . branching or
    # nupic . hypersearch . min . field . contribution was set ) , then be more
    # picky about which fields we add in .
    limitFields = False
    if self . _hsObj . _maxBranching > 0 or self . _hsObj . _minFieldContribution >= 0 :
        if self . _hsObj . _searchType == HsSearchType . temporal or self . _hsObj . _searchType == HsSearchType . classification :
            if sprintIdx >= 1 :
                limitFields = True
                baseSprintIdx = 0
        elif self . _hsObj . _searchType == HsSearchType . legacyTemporal :
            if sprintIdx >= 2 :
                limitFields = True
                baseSprintIdx = 1
        else :
            raise RuntimeError ( "Unimplemented search type %s" % ( self . _hsObj . _searchType ) )
    # Only add top _ maxBranching encoders to the swarms ?
    if limitFields : # Get field contributions to filter added fields
        pctFieldContributions , absFieldContributions = self . getFieldContributions ( )
        toRemove = [ ]
        self . logger . debug ( "FieldContributions min: %s" % ( self . _hsObj . _minFieldContribution ) )
        for fieldname in pctFieldContributions :
            if pctFieldContributions [ fieldname ] < self . _hsObj . _minFieldContribution :
                self . logger . debug ( "FieldContributions removing: %s" % ( fieldname ) )
                toRemove . append ( self . getEncoderKeyFromName ( fieldname ) )
            else :
                self . logger . debug ( "FieldContributions keeping: %s" % ( fieldname ) )
        # Grab the top maxBranching base sprint swarms .
        swarms = self . _state [ "swarms" ]
        sprintSwarms = [ ( swarm , swarms [ swarm ] [ "bestErrScore" ] ) for swarm in swarms if swarms [ swarm ] [ "sprintIdx" ] == baseSprintIdx ]
        sprintSwarms = sorted ( sprintSwarms , key = itemgetter ( 1 ) )
        if self . _hsObj . _maxBranching > 0 :
            sprintSwarms = sprintSwarms [ 0 : self . _hsObj . _maxBranching ]
        # Create encoder set to generate further swarms .
        for swarm in sprintSwarms :
            swarmEncoders = swarm [ 0 ] . split ( "." )
            for encoder in swarmEncoders :
                if not encoder in encoderAddSet :
                    encoderAddSet . append ( encoder )
        encoderAddSet = [ encoder for encoder in encoderAddSet if not str ( encoder ) in toRemove ]
    # If no limit on the branching or min contribution , simply use all of the
    # encoders .
    else :
        encoderAddSet = self . _hsObj . _encoderNames
    # Build up the new encoder combinations for the next sprint .
    newSwarmIds = set ( )
    # See if the caller wants to try more extensive field combinations with
    # 3 fields .
    if ( self . _hsObj . _searchType == HsSearchType . temporal or self . _hsObj . _searchType == HsSearchType . legacyTemporal ) and sprintIdx == 2 and ( self . _hsObj . _tryAll3FieldCombinations or self . _hsObj . _tryAll3FieldCombinationsWTimestamps ) :
        if self . _hsObj . _tryAll3FieldCombinations :
            newEncoders = set ( self . _hsObj . _encoderNames )
            if self . _hsObj . _predictedFieldEncoder in newEncoders :
                newEncoders . remove ( self . _hsObj . _predictedFieldEncoder )
        else : # Just make sure the timestamp encoders are part of the mix
            newEncoders = set ( encoderAddSet )
            if self . _hsObj . _predictedFieldEncoder in newEncoders :
                newEncoders . remove ( self . _hsObj . _predictedFieldEncoder )
            for encoder in self . _hsObj . _encoderNames :
                if encoder . endswith ( '_timeOfDay' ) or encoder . endswith ( '_weekend' ) or encoder . endswith ( '_dayOfWeek' ) :
                    newEncoders . add ( encoder )
        allCombos = list ( itertools . combinations ( newEncoders , 2 ) )
        for combo in allCombos :
            newSet = list ( combo )
            newSet . append ( self . _hsObj . _predictedFieldEncoder )
            newSet . sort ( )
            newSwarmId = '.' . join ( newSet )
            if newSwarmId not in self . _state [ 'swarms' ] :
                newSwarmIds . add ( newSwarmId )
                # If a speculative sprint , only add the first encoder , if not add
                # all of them .
                if ( len ( self . getActiveSwarms ( sprintIdx - 1 ) ) > 0 ) :
                    break
    # Else , we only build up by adding 1 new encoder to the best combination ( s )
    # we ' ve seen from the prior sprint
    else :
        for baseEncoderSet in baseEncoderSets :
            for encoder in encoderAddSet :
                if encoder not in self . _state [ 'blackListedEncoders' ] and encoder not in baseEncoderSet :
                    newSet = list ( baseEncoderSet )
                    newSet . append ( encoder )
                    newSet . sort ( )
                    newSwarmId = '.' . join ( newSet )
                    if newSwarmId not in self . _state [ 'swarms' ] :
                        newSwarmIds . add ( newSwarmId )
                        # If a speculative sprint , only add the first encoder , if not add
                        # all of them .
                        if ( len ( self . getActiveSwarms ( sprintIdx - 1 ) ) > 0 ) :
                            break
    # Sort the new swarm Ids
    newSwarmIds = sorted ( newSwarmIds )
    # If no more swarms can be found for this sprint . . .
    if len ( newSwarmIds ) == 0 : # if sprint is not an empty sprint return that it is active but do not
    # add anything to it .
        if len ( self . getAllSwarms ( sprintIdx ) ) > 0 :
            return ( True , False )
        # If this is an empty sprint and we couldn ' t find any new swarms to
        # add ( only bad fields are remaining ) , the search is over
        else :
            return ( False , True )
    # Add this sprint and the swarms that are in it to our state
    self . _dirty = True
    # Add in the new sprint if necessary
    if len ( self . _state [ "sprints" ] ) == sprintIdx :
        self . _state [ 'sprints' ] . append ( { 'status' : 'active' , 'bestModelId' : None , 'bestErrScore' : None } )
    # Add in the new swarm ( s ) to the sprint
    for swarmId in newSwarmIds :
        self . _state [ 'swarms' ] [ swarmId ] = { 'status' : 'active' , 'bestModelId' : None , 'bestErrScore' : None , 'sprintIdx' : sprintIdx }
    # Update the list of active swarms
    self . _state [ 'activeSwarms' ] = self . getActiveSwarms ( )
    # Try to set new state
    success = self . writeStateToDB ( )
    # Return result if successful
    if success :
        return ( True , False )


signature:
def addEncoder ( self , name , encoder ) :
    """Adds one encoder .
 : param name : ( string ) name of encoder , should be unique
 : param encoder : ( : class : ` . Encoder ` ) the encoder to add"""

implementation:
self . encoders . append ( ( name , encoder , self . width ) )
for d in encoder . getDescription ( ) :
    self . description . append ( ( d [ 0 ] , d [ 1 ] + self . width ) )
self . width += encoder . getWidth ( )


signature:
def invariant ( self ) :
    """Verify the validity of the node spec object
 The type of each sub - object is verified and then
 the validity of each node spec item is verified by calling
 it invariant ( ) method . It also makes sure that there is at most
 one default input and one default output ."""

implementation:
# Verify the description and singleNodeOnly attributes
assert isinstance ( self . description , str )
assert isinstance ( self . singleNodeOnly , bool )
# Make sure that all items dicts are really dicts
assert isinstance ( self . inputs , dict )
assert isinstance ( self . outputs , dict )
assert isinstance ( self . parameters , dict )
assert isinstance ( self . commands , dict )
# Verify all item dicts
hasDefaultInput = False
for k , v in self . inputs . items ( ) :
    assert isinstance ( k , str )
    assert isinstance ( v , InputSpec )
    v . invariant ( )
    if v . isDefaultInput :
        assert not hasDefaultInput
        hasDefaultInput = True
hasDefaultOutput = False
for k , v in self . outputs . items ( ) :
    assert isinstance ( k , str )
    assert isinstance ( v , OutputSpec )
    v . invariant ( )
    if v . isDefaultOutput :
        assert not hasDefaultOutput
        hasDefaultOutput = True
for k , v in self . parameters . items ( ) :
    assert isinstance ( k , str )
    assert isinstance ( v , ParameterSpec )
    v . invariant ( )
for k , v in self . commands . items ( ) :
    assert isinstance ( k , str )
    assert isinstance ( v , CommandSpec )
    v . invariant ( )


signature:
def updateResultsForJob ( self , forceUpdate = True ) :
    """Chooses the best model for a given job .
 Parameters
 forceUpdate : ( True / False ) . If True , the update will ignore all the
 restrictions on the minimum time to update and the minimum
 number of records to update . This should typically only be
 set to true if the model has completed running"""

implementation:
updateInterval = time . time ( ) - self . _lastUpdateAttemptTime
if updateInterval < self . _MIN_UPDATE_INTERVAL and not forceUpdate :
    return
self . logger . info ( "Attempting model selection for jobID=%d: time=%f" "  lastUpdate=%f" % ( self . _jobID , time . time ( ) , self . _lastUpdateAttemptTime ) )
timestampUpdated = self . _cjDB . jobUpdateSelectionSweep ( self . _jobID , self . _MIN_UPDATE_INTERVAL )
if not timestampUpdated :
    self . logger . info ( "Unable to update selection sweep timestamp: jobID=%d" " updateTime=%f" % ( self . _jobID , self . _lastUpdateAttemptTime ) )
    if not forceUpdate :
        return
self . _lastUpdateAttemptTime = time . time ( )
self . logger . info ( "Succesfully updated selection sweep timestamp jobid=%d updateTime=%f" % ( self . _jobID , self . _lastUpdateAttemptTime ) )
minUpdateRecords = self . _MIN_UPDATE_THRESHOLD
jobResults = self . _getJobResults ( )
if forceUpdate or jobResults is None :
    minUpdateRecords = 0
candidateIDs , bestMetric = self . _cjDB . modelsGetCandidates ( self . _jobID , minUpdateRecords )
self . logger . info ( "Candidate models=%s, metric=%s, jobID=%s" % ( candidateIDs , bestMetric , self . _jobID ) )
if len ( candidateIDs ) == 0 :
    return
self . _jobUpdateCandidate ( candidateIDs [ 0 ] , bestMetric , results = jobResults )


signature:
def createEncoder ( ) :
    """Create the encoder instance for our test and return it ."""

implementation:
consumption_encoder = ScalarEncoder ( 21 , 0.0 , 100.0 , n = 50 , name = "consumption" , clipInput = True )
time_encoder = DateEncoder ( timeOfDay = ( 21 , 9.5 ) , name = "timestamp_timeOfDay" )
encoder = MultiEncoder ( )
encoder . addEncoder ( "consumption" , consumption_encoder )
encoder . addEncoder ( "timestamp" , time_encoder )
return encoder


signature:
def createNetwork ( dataSource ) :
    """Create the Network instance .
 The network has a sensor region reading data from ` dataSource ` and passing
 the encoded representation to an SPRegion . The SPRegion output is passed to
 a TMRegion .
 : param dataSource : a RecordStream instance to get data from
 : returns : a Network instance ready to run"""

implementation:
network = Network ( )
# Our input is sensor data from the gym file . The RecordSensor region
# allows us to specify a file record stream as the input source via the
# dataSource attribute .
network . addRegion ( "sensor" , "py.RecordSensor" , json . dumps ( { "verbosity" : _VERBOSITY } ) )
sensor = network . regions [ "sensor" ] . getSelf ( )
# The RecordSensor needs to know how to encode the input values
sensor . encoder = createEncoder ( )
# Specify the dataSource as a file record stream instance
sensor . dataSource = dataSource
# Create the spatial pooler region
SP_PARAMS [ "inputWidth" ] = sensor . encoder . getWidth ( )
network . addRegion ( "spatialPoolerRegion" , "py.SPRegion" , json . dumps ( SP_PARAMS ) )
# Link the SP region to the sensor input
network . link ( "sensor" , "spatialPoolerRegion" , "UniformLink" , "" )
network . link ( "sensor" , "spatialPoolerRegion" , "UniformLink" , "" , srcOutput = "resetOut" , destInput = "resetIn" )
network . link ( "spatialPoolerRegion" , "sensor" , "UniformLink" , "" , srcOutput = "spatialTopDownOut" , destInput = "spatialTopDownIn" )
network . link ( "spatialPoolerRegion" , "sensor" , "UniformLink" , "" , srcOutput = "temporalTopDownOut" , destInput = "temporalTopDownIn" )
# Add the TMRegion on top of the SPRegion
network . addRegion ( "temporalPoolerRegion" , "py.TMRegion" , json . dumps ( TM_PARAMS ) )
network . link ( "spatialPoolerRegion" , "temporalPoolerRegion" , "UniformLink" , "" )
network . link ( "temporalPoolerRegion" , "spatialPoolerRegion" , "UniformLink" , "" , srcOutput = "topDownOut" , destInput = "topDownIn" )
# Add the AnomalyLikelihoodRegion on top of the TMRegion
network . addRegion ( "anomalyLikelihoodRegion" , "py.AnomalyLikelihoodRegion" , json . dumps ( { } ) )
network . link ( "temporalPoolerRegion" , "anomalyLikelihoodRegion" , "UniformLink" , "" , srcOutput = "anomalyScore" , destInput = "rawAnomalyScore" )
network . link ( "sensor" , "anomalyLikelihoodRegion" , "UniformLink" , "" , srcOutput = "sourceOut" , destInput = "metricValue" )
spatialPoolerRegion = network . regions [ "spatialPoolerRegion" ]
# Make sure learning is enabled
spatialPoolerRegion . setParameter ( "learningMode" , True )
# We want temporal anomalies so disable anomalyMode in the SP . This mode is
# used for computing anomalies in a non - temporal model .
spatialPoolerRegion . setParameter ( "anomalyMode" , False )
temporalPoolerRegion = network . regions [ "temporalPoolerRegion" ]
# Enable topDownMode to get the predicted columns output
temporalPoolerRegion . setParameter ( "topDownMode" , True )
# Make sure learning is enabled ( this is the default )
temporalPoolerRegion . setParameter ( "learningMode" , True )
# Enable inference mode so we get predictions
temporalPoolerRegion . setParameter ( "inferenceMode" , True )
# Enable anomalyMode to compute the anomaly score to be passed to the anomaly
# likelihood region .
temporalPoolerRegion . setParameter ( "anomalyMode" , True )
return network


signature:
def runNetwork ( network , writer ) :
    """Run the network and write output to writer .
 : param network : a Network instance to run
 : param writer : a csv . writer instance to write output to"""

implementation:
sensorRegion = network . regions [ "sensor" ]
spatialPoolerRegion = network . regions [ "spatialPoolerRegion" ]
temporalPoolerRegion = network . regions [ "temporalPoolerRegion" ]
anomalyLikelihoodRegion = network . regions [ "anomalyLikelihoodRegion" ]
prevPredictedColumns = [ ]
for i in xrange ( _NUM_RECORDS ) : # Run the network for a single iteration
    network . run ( 1 )
    # Write out the anomaly likelihood along with the record number and consumption
    # value .
    consumption = sensorRegion . getOutputData ( "sourceOut" ) [ 0 ]
    anomalyScore = temporalPoolerRegion . getOutputData ( "anomalyScore" ) [ 0 ]
    anomalyLikelihood = anomalyLikelihoodRegion . getOutputData ( "anomalyLikelihood" ) [ 0 ]
    writer . writerow ( ( i , consumption , anomalyScore , anomalyLikelihood ) )


signature:
def __validateExperimentControl ( self , control ) :
    """Validates control dictionary for the experiment context"""

implementation:
# Validate task list
taskList = control . get ( 'tasks' , None )
if taskList is not None :
    taskLabelsList = [ ]
    for task in taskList :
        validateOpfJsonValue ( task , "opfTaskSchema.json" )
        validateOpfJsonValue ( task [ 'taskControl' ] , "opfTaskControlSchema.json" )
        taskLabel = task [ 'taskLabel' ]
        assert isinstance ( taskLabel , types . StringTypes ) , "taskLabel type: %r" % type ( taskLabel )
        assert len ( taskLabel ) > 0 , "empty string taskLabel not is allowed"
        taskLabelsList . append ( taskLabel . lower ( ) )
    taskLabelDuplicates = filter ( lambda x : taskLabelsList . count ( x ) > 1 , taskLabelsList )
    assert len ( taskLabelDuplicates ) == 0 , "Duplcate task labels are not allowed: %s" % taskLabelDuplicates
return


signature:
def normalizeStreamSource ( self , stream ) :
    """TODO : document
 : param stream :"""

implementation:
# The stream source in the task might be in several formats , so we need
# to make sure it gets converted into an absolute path :
source = stream [ 'source' ] [ len ( FILE_SCHEME ) : ]
# If the source is already an absolute path , we won ' t use pkg _ resources ,
# we ' ll just trust that the path exists . If not , it ' s a user problem .
if os . path . isabs ( source ) :
    sourcePath = source
else : # First we ' ll check to see if this path exists within the nupic . datafiles
# package resource .
    sourcePath = resource_filename ( "nupic.datafiles" , source )
    if not os . path . exists ( sourcePath ) : # If this path doesn ' t exist as a package resource , the last option will
    # be to treat it as a relative path to the current working directory .
        sourcePath = os . path . join ( os . getcwd ( ) , source )
stream [ 'source' ] = FILE_SCHEME + sourcePath


signature:
def normalizeStreamSources ( self ) :
    """TODO : document"""

implementation:
task = dict ( self . __control )
if 'dataset' in task :
    for stream in task [ 'dataset' ] [ 'streams' ] :
        self . normalizeStreamSource ( stream )
else :
    for subtask in task [ 'tasks' ] :
        for stream in subtask [ 'dataset' ] [ 'streams' ] :
            self . normalizeStreamSource ( stream )


signature:
def convertNupicEnvToOPF ( self ) :
    """TODO : document"""

implementation:
# We need to create a task structure , most of which is taken verbatim
# from the Nupic control dict
task = dict ( self . __control )
task . pop ( 'environment' )
inferenceArgs = task . pop ( 'inferenceArgs' )
task [ 'taskLabel' ] = 'DefaultTask'
# Create the iterationCycle element that will be placed inside the
# taskControl .
iterationCount = task . get ( 'iterationCount' , - 1 )
iterationCountInferOnly = task . pop ( 'iterationCountInferOnly' , 0 )
if iterationCountInferOnly == - 1 :
    iterationCycle = [ IterationPhaseSpecInferOnly ( 1000 , inferenceArgs = inferenceArgs ) ]
elif iterationCountInferOnly > 0 :
    assert iterationCount > 0 , "When iterationCountInferOnly is specified, " "iterationCount must also be specified and not be -1"
    iterationCycle = [ IterationPhaseSpecLearnAndInfer ( iterationCount - iterationCountInferOnly , inferenceArgs = inferenceArgs ) , IterationPhaseSpecInferOnly ( iterationCountInferOnly , inferenceArgs = inferenceArgs ) ]
else :
    iterationCycle = [ IterationPhaseSpecLearnAndInfer ( 1000 , inferenceArgs = inferenceArgs ) ]
taskControl = dict ( metrics = task . pop ( 'metrics' ) , loggedMetrics = task . pop ( 'loggedMetrics' ) , iterationCycle = iterationCycle )
task [ 'taskControl' ] = taskControl
# Create the new control
self . __control = dict ( environment = OpfEnvironment . Nupic , tasks = [ task ] )


signature:
def createNetwork ( dataSource ) :
    """Create the Network instance .
 The network has a sensor region reading data from ` dataSource ` and passing
 the encoded representation to an Identity Region .
 : param dataSource : a RecordStream instance to get data from
 : returns : a Network instance ready to run"""

implementation:
network = Network ( )
# Our input is sensor data from the gym file . The RecordSensor region
# allows us to specify a file record stream as the input source via the
# dataSource attribute .
network . addRegion ( "sensor" , "py.RecordSensor" , json . dumps ( { "verbosity" : _VERBOSITY } ) )
sensor = network . regions [ "sensor" ] . getSelf ( )
# The RecordSensor needs to know how to encode the input values
sensor . encoder = createEncoder ( )
# Specify the dataSource as a file record stream instance
sensor . dataSource = dataSource
# CUSTOM REGION
# Add path to custom region to PYTHONPATH
# NOTE : Before using a custom region , please modify your PYTHONPATH
# export PYTHONPATH = " < path to custom region module > : $ PYTHONPATH "
# In this demo , we have modified it using sys . path . append since we need it to
# have an effect on this program .
sys . path . append ( os . path . dirname ( os . path . abspath ( __file__ ) ) )
from custom_region . identity_region import IdentityRegion
# Add custom region class to the network
Network . registerRegion ( IdentityRegion )
# Create a custom region
network . addRegion ( "identityRegion" , "py.IdentityRegion" , json . dumps ( { "dataWidth" : sensor . encoder . getWidth ( ) , } ) )
# Link the Identity region to the sensor output
network . link ( "sensor" , "identityRegion" , "UniformLink" , "" )
network . initialize ( )
return network


implementation:
identityRegion = network . regions [ "identityRegion" ]
for i in xrange ( _NUM_RECORDS ) : # Run the network for a single iteration
    network . run ( 1 )
    # Write out the record number and encoding
    encoding = identityRegion . getOutputData ( "out" )
    writer . writerow ( ( i , encoding ) )


signature:
def _appendReportKeys ( keys , prefix , results ) :
    """Generate a set of possible report keys for an experiment ' s results .
 A report key is a string of key names separated by colons , each key being one
 level deeper into the experiment results dict . For example , ' key1 : key2 ' .
 This routine is called recursively to build keys that are multiple levels
 deep from the results dict .
 Parameters :
 keys : Set of report keys accumulated so far
 prefix : prefix formed so far , this is the colon separated list of key
 names that led up to the dict passed in results
 results : dictionary of results at this level ."""

implementation:
allKeys = results . keys ( )
allKeys . sort ( )
for key in allKeys :
    if hasattr ( results [ key ] , 'keys' ) :
        _appendReportKeys ( keys , "%s%s:" % ( prefix , key ) , results [ key ] )
    else :
        keys . add ( "%s%s" % ( prefix , key ) )


signature:
def _matchReportKeys ( reportKeyREs = [ ] , allReportKeys = [ ] ) :
    """Extract all items from the ' allKeys ' list whose key matches one of the regular
 expressions passed in ' reportKeys ' .
 Parameters :
 reportKeyREs : List of regular expressions
 allReportKeys : List of all keys
 retval : list of keys from allReportKeys that match the regular expressions
 in ' reportKeyREs '
 If an invalid regular expression was included in ' reportKeys ' ,
 then BadKeyError ( ) is raised"""

implementation:
matchingReportKeys = [ ]
# Extract the report items of interest
for keyRE in reportKeyREs : # Find all keys that match this regular expression
    matchObj = re . compile ( keyRE )
    found = False
    for keyName in allReportKeys :
        match = matchObj . match ( keyName )
        if match and match . end ( ) == len ( keyName ) :
            matchingReportKeys . append ( keyName )
            found = True
    if not found :
        raise _BadKeyError ( keyRE )
return matchingReportKeys


signature:
def _getReportItem ( itemName , results ) :
    """Get a specific item by name out of the results dict .
 The format of itemName is a string of dictionary keys separated by colons ,
 each key being one level deeper into the results dict . For example ,
 ' key1 : key2 ' would fetch results [ ' key1 ' ] [ ' key2 ' ] .
 If itemName is not found in results , then None is returned"""

implementation:
subKeys = itemName . split ( ':' )
subResults = results
for subKey in subKeys :
    subResults = subResults [ subKey ]
return subResults


signature:
def filterResults ( allResults , reportKeys , optimizeKey = None ) :
    """Given the complete set of results generated by an experiment ( passed in
 ' results ' ) , filter out and return only the ones the caller wants , as
 specified through ' reportKeys ' and ' optimizeKey ' .
 A report key is a string of key names separated by colons , each key being one
 level deeper into the experiment results dict . For example , ' key1 : key2 ' .
 Parameters :
 results : dict of all results generated by an experiment
 reportKeys : list of items from the results dict to include in
 the report . These can be regular expressions .
 optimizeKey : Which report item , if any , we will be optimizing for . This can
 also be a regular expression , but is an error if it matches
 more than one key from the experiment ' s results .
 retval : ( reportDict , optimizeDict )
 reportDict : a dictionary of the metrics named by desiredReportKeys
 optimizeDict : A dictionary containing 1 item : the full name and
 value of the metric identified by the optimizeKey"""

implementation:
# Init return values
optimizeDict = dict ( )
# Get all available report key names for this experiment
allReportKeys = set ( )
_appendReportKeys ( keys = allReportKeys , prefix = '' , results = allResults )
# Extract the report items that match the regular expressions passed in reportKeys
matchingKeys = _matchReportKeys ( reportKeys , allReportKeys )
# Extract the values of the desired items
reportDict = dict ( )
for keyName in matchingKeys :
    value = _getReportItem ( keyName , allResults )
    reportDict [ keyName ] = value
# Extract the report item that matches the regular expression passed in
# optimizeKey
if optimizeKey is not None :
    matchingKeys = _matchReportKeys ( [ optimizeKey ] , allReportKeys )
    if len ( matchingKeys ) == 0 :
        raise _BadKeyError ( optimizeKey )
    elif len ( matchingKeys ) > 1 :
        raise _BadOptimizeKeyError ( optimizeKey , matchingKeys )
    optimizeKeyFullName = matchingKeys [ 0 ]
    # Get the value of the optimize metric
    value = _getReportItem ( optimizeKeyFullName , allResults )
    optimizeDict [ optimizeKeyFullName ] = value
    reportDict [ optimizeKeyFullName ] = value
# Return info
return ( reportDict , optimizeDict )


signature:
def _handleModelRunnerException ( jobID , modelID , jobsDAO , experimentDir , logger , e ) :
    """Perform standard handling of an exception that occurs while running
 a model .
 Parameters :
 jobID : ID for this hypersearch job in the jobs table
 modelID : model ID
 jobsDAO : ClientJobsDAO instance
 experimentDir : directory containing the experiment
 logger : the logger to use
 e : the exception that occurred
 retval : ( completionReason , completionMsg )"""

implementation:
msg = StringIO . StringIO ( )
print >> msg , "Exception occurred while running model %s: %r (%s)" % ( modelID , e , type ( e ) )
traceback . print_exc ( None , msg )
completionReason = jobsDAO . CMPL_REASON_ERROR
completionMsg = msg . getvalue ( )
logger . error ( completionMsg )
# Write results to the model database for the error case . Ignore
# InvalidConnectionException , as this is usually caused by orphaned models
# TODO : do we really want to set numRecords to 0 ? Last updated value might
# be useful for debugging
if type ( e ) is not InvalidConnectionException :
    jobsDAO . modelUpdateResults ( modelID , results = None , numRecords = 0 )
# TODO : Make sure this wasn ' t the best model in job . If so , set the best
# appropriately
# If this was an exception that should mark the job as failed , do that
# now .
if type ( e ) == JobFailException :
    workerCmpReason = jobsDAO . jobGetFields ( jobID , [ 'workerCompletionReason' ] ) [ 0 ]
    if workerCmpReason == ClientJobsDAO . CMPL_REASON_SUCCESS :
        jobsDAO . jobSetFields ( jobID , fields = dict ( cancel = True , workerCompletionReason = ClientJobsDAO . CMPL_REASON_ERROR , workerCompletionMsg = ": " . join ( str ( i ) for i in e . args ) ) , useConnectionID = False , ignoreUnchanged = True )
return ( completionReason , completionMsg )