signature:
def _hsStatePeriodicUpdate ( self , exhaustedSwarmId = None ) :
    """Periodically , check to see if we should remove a certain field combination
 from evaluation ( because it is doing so poorly ) or move on to the next
 sprint ( add in more fields ) .
 This method is called from _ getCandidateParticleAndSwarm ( ) , which is called
 right before we try and create a new model to run .
 Parameters :
 removeSwarmId : If not None , force a change to the current set of active
 swarms by removing this swarm . This is used in situations
 where we can ' t find any new unique models to create in
 this swarm . In these situations , we update the hypersearch
 state regardless of the timestamp of the last time another
 worker updated it ."""

implementation:
if self . _hsState is None :
    self . _hsState = HsState ( self )
# Read in current state from the DB
self . _hsState . readStateFromDB ( )
# This will hold the list of completed swarms that we find
completedSwarms = set ( )
# Mark the exhausted swarm as completing / completed , if any
if exhaustedSwarmId is not None :
    self . logger . info ( "Removing swarm %s from the active set " "because we can't find any new unique particle " "positions" % ( exhaustedSwarmId ) )
    # Is it completing or completed ?
    ( particles , _ , _ , _ , _ ) = self . _resultsDB . getParticleInfos ( swarmId = exhaustedSwarmId , matured = False )
    if len ( particles ) > 0 :
        exhaustedSwarmStatus = 'completing'
    else :
        exhaustedSwarmStatus = 'completed'
# Kill all swarms that don ' t need to be explored based on the most recent
# information .
if self . _killUselessSwarms :
    self . _hsState . killUselessSwarms ( )
# For all swarms that were in the ' completing ' state , see if they have
# completed yet .
# Note that we are not quite sure why this doesn ' t automatically get handled
# when we receive notification that a model finally completed in a swarm .
# But , we ARE running into a situation , when speculativeParticles is off ,
# where we have one or more swarms in the ' completing ' state even though all
# models have since finished . This logic will serve as a failsafe against
# this situation .
completingSwarms = self . _hsState . getCompletingSwarms ( )
for swarmId in completingSwarms : # Is it completed ?
    ( particles , _ , _ , _ , _ ) = self . _resultsDB . getParticleInfos ( swarmId = swarmId , matured = False )
    if len ( particles ) == 0 :
        completedSwarms . add ( swarmId )
# Are there any swarms we can remove ( because they have matured ) ?
completedSwarmGens = self . _resultsDB . getMaturedSwarmGenerations ( )
priorCompletedSwarms = self . _hsState . getCompletedSwarms ( )
for ( swarmId , genIdx , errScore ) in completedSwarmGens : # Don ' t need to report it if the swarm already completed
    if swarmId in priorCompletedSwarms :
        continue
    completedList = self . _swarmTerminator . recordDataPoint ( swarmId = swarmId , generation = genIdx , errScore = errScore )
    # Update status message
    statusMsg = "Completed generation #%d of swarm '%s' with a best" " errScore of %g" % ( genIdx , swarmId , errScore )
    if len ( completedList ) > 0 :
        statusMsg = "%s. Matured swarm(s): %s" % ( statusMsg , completedList )
    self . logger . info ( statusMsg )
    self . _cjDAO . jobSetFields ( jobID = self . _jobID , fields = dict ( engStatus = statusMsg ) , useConnectionID = False , ignoreUnchanged = True )
    # Special test mode to check which swarms have terminated
    if 'NTA_TEST_recordSwarmTerminations' in os . environ :
        while True :
            resultsStr = self . _cjDAO . jobGetFields ( self . _jobID , [ 'results' ] ) [ 0 ]
            if resultsStr is None :
                results = { }
            else :
                results = json . loads ( resultsStr )
            if not 'terminatedSwarms' in results :
                results [ 'terminatedSwarms' ] = { }
            for swarm in completedList :
                if swarm not in results [ 'terminatedSwarms' ] :
                    results [ 'terminatedSwarms' ] [ swarm ] = ( genIdx , self . _swarmTerminator . swarmScores [ swarm ] )
            newResultsStr = json . dumps ( results )
            if newResultsStr == resultsStr :
                break
            updated = self . _cjDAO . jobSetFieldIfEqual ( jobID = self . _jobID , fieldName = 'results' , curValue = resultsStr , newValue = json . dumps ( results ) )
            if updated :
                break
    if len ( completedList ) > 0 :
        for name in completedList :
            self . logger . info ( "Swarm matured: %s. Score at generation %d: " "%s" % ( name , genIdx , errScore ) )
        completedSwarms = completedSwarms . union ( completedList )
if len ( completedSwarms ) == 0 and ( exhaustedSwarmId is None ) :
    return
# We need to mark one or more swarms as completed , keep trying until
# successful , or until some other worker does it for us .
while True :
    if exhaustedSwarmId is not None :
        self . _hsState . setSwarmState ( exhaustedSwarmId , exhaustedSwarmStatus )
    # Mark the completed swarms as completed
    for swarmId in completedSwarms :
        self . _hsState . setSwarmState ( swarmId , 'completed' )
    # If nothing changed , we ' re done
    if not self . _hsState . isDirty ( ) :
        return
    # Update the shared Hypersearch state now
    # This will do nothing and return False if some other worker beat us to it
    success = self . _hsState . writeStateToDB ( )
    if success : # Go through and cancel all models that are still running , except for
    # the best model . Once the best model changes , the one that used to be
    # best ( and has matured ) will notice that and stop itself at that point .
        jobResultsStr = self . _cjDAO . jobGetFields ( self . _jobID , [ 'results' ] ) [ 0 ]
        if jobResultsStr is not None :
            jobResults = json . loads ( jobResultsStr )
            bestModelId = jobResults . get ( 'bestModel' , None )
        else :
            bestModelId = None
        for swarmId in list ( completedSwarms ) :
            ( _ , modelIds , _ , _ , _ ) = self . _resultsDB . getParticleInfos ( swarmId = swarmId , completed = False )
            if bestModelId in modelIds :
                modelIds . remove ( bestModelId )
            if len ( modelIds ) == 0 :
                continue
            self . logger . info ( "Killing the following models in swarm '%s' because" "the swarm is being terminated: %s" % ( swarmId , str ( modelIds ) ) )
            for modelId in modelIds :
                self . _cjDAO . modelSetFields ( modelId , dict ( engStop = ClientJobsDAO . STOP_REASON_KILLED ) , ignoreUnchanged = True )
        return
    # We were not able to change the state because some other worker beat us
    # to it .
    # Get the new state , and try again to apply our changes .
    self . _hsState . readStateFromDB ( )
    self . logger . debug ( "New hsState has been set by some other worker to: " " \n%s" % ( pprint . pformat ( self . _hsState . _state , indent = 4 ) ) )


signature:
def _getCandidateParticleAndSwarm ( self , exhaustedSwarmId = None ) :
    """Find or create a candidate particle to produce a new model .
 At any one time , there is an active set of swarms in the current sprint , where
 each swarm in the sprint represents a particular combination of fields .
 Ideally , we should try to balance the number of models we have evaluated for
 each swarm at any time .
 This method will see how many models have been evaluated for each active
 swarm in the current active sprint ( s ) and then try and choose a particle
 from the least represented swarm in the first possible active sprint , with
 the following constraints / rules :
 for each active sprint :
 for each active swarm ( preference to those with least # of models so far ) :
 1 . ) The particle will be created from new ( generation # 0 ) if there are not
 already self . _ minParticlesPerSwarm particles in the swarm .
 2 . ) Find the first gen that has a completed particle and evolve that
 particle to the next generation .
 3 . ) If we got to here , we know that we have satisfied the min # of
 particles for the swarm , and they are all currently running ( probably at
 various generation indexes ) . Go onto the next swarm
 If we couldn ' t find a swarm to allocate a particle in , go onto the next
 sprint and start allocating particles there . . . .
 Parameters :
 exhaustedSwarmId : If not None , force a change to the current set of active
 swarms by marking this swarm as either ' completing ' or
 ' completed ' . If there are still models being evaluaed in
 it , mark it as ' completing ' , else ' completed . This is
 used in situations where we can ' t find any new unique
 models to create in this swarm . In these situations , we
 force an update to the hypersearch state so no other
 worker wastes time try to use this swarm .
 retval : ( exit , particle , swarm )
 exit : If true , this worker is ready to exit ( particle and
 swarm will be None )
 particle : Which particle to run
 swarm : which swarm the particle is in
 NOTE : When particle and swarm are None and exit is False , it
 means that we need to wait for one or more other worker ( s ) to
 finish their respective models before we can pick a particle
 to run . This will generally only happen when speculativeParticles
 is set to False ."""

implementation:
# Cancel search ?
jobCancel = self . _cjDAO . jobGetFields ( self . _jobID , [ 'cancel' ] ) [ 0 ]
if jobCancel :
    self . _jobCancelled = True
    # Did a worker cancel the job because of an error ?
    ( workerCmpReason , workerCmpMsg ) = self . _cjDAO . jobGetFields ( self . _jobID , [ 'workerCompletionReason' , 'workerCompletionMsg' ] )
    if workerCmpReason == ClientJobsDAO . CMPL_REASON_SUCCESS :
        self . logger . info ( "Exiting due to job being cancelled" )
        self . _cjDAO . jobSetFields ( self . _jobID , dict ( workerCompletionMsg = "Job was cancelled" ) , useConnectionID = False , ignoreUnchanged = True )
    else :
        self . logger . error ( "Exiting because some worker set the " "workerCompletionReason to %s. WorkerCompletionMsg: %s" % ( workerCmpReason , workerCmpMsg ) )
    return ( True , None , None )
# Perform periodic updates on the Hypersearch state .
if self . _hsState is not None :
    priorActiveSwarms = self . _hsState . getActiveSwarms ( )
else :
    priorActiveSwarms = None
# Update the HypersearchState , checking for matured swarms , and marking
# the passed in swarm as exhausted , if any
self . _hsStatePeriodicUpdate ( exhaustedSwarmId = exhaustedSwarmId )
# The above call may have modified self . _ hsState [ ' activeSwarmIds ' ]
# Log the current set of active swarms
activeSwarms = self . _hsState . getActiveSwarms ( )
if activeSwarms != priorActiveSwarms :
    self . logger . info ( "Active swarms changed to %s (from %s)" % ( activeSwarms , priorActiveSwarms ) )
self . logger . debug ( "Active swarms: %s" % ( activeSwarms ) )
# If too many model errors were detected , exit
totalCmpModels = self . _resultsDB . getNumCompletedModels ( )
if totalCmpModels > 5 :
    numErrs = self . _resultsDB . getNumErrModels ( )
    if ( float ( numErrs ) / totalCmpModels ) > self . _maxPctErrModels : # Get one of the errors
        errModelIds = self . _resultsDB . getErrModelIds ( )
        resInfo = self . _cjDAO . modelsGetResultAndStatus ( [ errModelIds [ 0 ] ] ) [ 0 ]
        modelErrMsg = resInfo . completionMsg
        cmpMsg = "%s: Exiting due to receiving too many models failing" " from exceptions (%d out of %d). \nModel Exception: %s" % ( ErrorCodes . tooManyModelErrs , numErrs , totalCmpModels , modelErrMsg )
        self . logger . error ( cmpMsg )
        # Cancel the entire job now , if it has not already been cancelled
        workerCmpReason = self . _cjDAO . jobGetFields ( self . _jobID , [ 'workerCompletionReason' ] ) [ 0 ]
        if workerCmpReason == ClientJobsDAO . CMPL_REASON_SUCCESS :
            self . _cjDAO . jobSetFields ( self . _jobID , fields = dict ( cancel = True , workerCompletionReason = ClientJobsDAO . CMPL_REASON_ERROR , workerCompletionMsg = cmpMsg ) , useConnectionID = False , ignoreUnchanged = True )
        return ( True , None , None )
# If HsState thinks the search is over , exit . It is seeing if the results
# on the sprint we just completed are worse than a prior sprint .
if self . _hsState . isSearchOver ( ) :
    cmpMsg = "Exiting because results did not improve in most recently" " completed sprint."
    self . logger . info ( cmpMsg )
    self . _cjDAO . jobSetFields ( self . _jobID , dict ( workerCompletionMsg = cmpMsg ) , useConnectionID = False , ignoreUnchanged = True )
    return ( True , None , None )
# Search successive active sprints , until we can find a candidate particle
# to work with
sprintIdx = - 1
while True : # Is this sprint active ?
    sprintIdx += 1
    ( active , eos ) = self . _hsState . isSprintActive ( sprintIdx )
    # If no more sprints to explore :
    if eos : # If any prior ones are still being explored , finish up exploring them
        if self . _hsState . anyGoodSprintsActive ( ) :
            self . logger . info ( "No more sprints to explore, waiting for prior" " sprints to complete" )
            return ( False , None , None )
        # Else , we ' re done
        else :
            cmpMsg = "Exiting because we've evaluated all possible field " "combinations"
            self . _cjDAO . jobSetFields ( self . _jobID , dict ( workerCompletionMsg = cmpMsg ) , useConnectionID = False , ignoreUnchanged = True )
            self . logger . info ( cmpMsg )
            return ( True , None , None )
    if not active :
        if not self . _speculativeParticles :
            if not self . _hsState . isSprintCompleted ( sprintIdx ) :
                self . logger . info ( "Waiting for all particles in sprint %d to complete" "before evolving any more particles" % ( sprintIdx ) )
                return ( False , None , None )
        continue
    # Look for swarms that have particle " holes " in their generations . That is ,
    # an earlier generation with less than minParticlesPerSwarm . This can
    # happen if a model that was started eariler got orphaned . If we detect
    # this , start a new particle in that generation .
    swarmIds = self . _hsState . getActiveSwarms ( sprintIdx )
    for swarmId in swarmIds :
        firstNonFullGenIdx = self . _resultsDB . firstNonFullGeneration ( swarmId = swarmId , minNumParticles = self . _minParticlesPerSwarm )
        if firstNonFullGenIdx is None :
            continue
        if firstNonFullGenIdx < self . _resultsDB . highestGeneration ( swarmId ) :
            self . logger . info ( "Cloning an earlier model in generation %d of swarm " "%s (sprintIdx=%s) to replace an orphaned model" % ( firstNonFullGenIdx , swarmId , sprintIdx ) )
            # Clone a random orphaned particle from the incomplete generation
            ( allParticles , allModelIds , errScores , completed , matured ) = self . _resultsDB . getOrphanParticleInfos ( swarmId , firstNonFullGenIdx )
            if len ( allModelIds ) > 0 : # We have seen instances where we get stuck in a loop incessantly
            # trying to clone earlier models ( NUP - 1511 ) . My best guess is that
            # we ' ve already successfully cloned each of the orphaned models at
            # least once , but still need at least one more . If we don ' t create
            # a new particleID , we will never be able to instantiate another
            # model ( since particleID hash is a unique key in the models table ) .
            # So , on 1/8/2013 this logic was changed to create a new particleID
            # whenever we clone an orphan .
                newParticleId = True
                self . logger . info ( "Cloning an orphaned model" )
            # If there is no orphan , clone one of the other particles . We can
            # have no orphan if this was a speculative generation that only
            # continued particles completed in the prior generation .
            else :
                newParticleId = True
                self . logger . info ( "No orphans found, so cloning a non-orphan" )
                ( allParticles , allModelIds , errScores , completed , matured ) = self . _resultsDB . getParticleInfos ( swarmId = swarmId , genIdx = firstNonFullGenIdx )
            # Clone that model
            modelId = random . choice ( allModelIds )
            self . logger . info ( "Cloning model %r" % ( modelId ) )
            ( particleState , _ , _ , _ , _ ) = self . _resultsDB . getParticleInfo ( modelId )
            particle = Particle ( hsObj = self , resultsDB = self . _resultsDB , flattenedPermuteVars = self . _flattenedPermutations , newFromClone = particleState , newParticleId = newParticleId )
            return ( False , particle , swarmId )
    # Sort the swarms in priority order , trying the ones with the least
    # number of models first
    swarmSizes = numpy . array ( [ self . _resultsDB . numModels ( x ) for x in swarmIds ] )
    swarmSizeAndIdList = zip ( swarmSizes , swarmIds )
    swarmSizeAndIdList . sort ( )
    for ( _ , swarmId ) in swarmSizeAndIdList : # 1 . ) The particle will be created from new ( at generation # 0 ) if there
    # are not already self . _ minParticlesPerSwarm particles in the swarm .
        ( allParticles , allModelIds , errScores , completed , matured ) = ( self . _resultsDB . getParticleInfos ( swarmId ) )
        if len ( allParticles ) < self . _minParticlesPerSwarm :
            particle = Particle ( hsObj = self , resultsDB = self . _resultsDB , flattenedPermuteVars = self . _flattenedPermutations , swarmId = swarmId , newFarFrom = allParticles )
            # Jam in the best encoder state found from the first sprint
            bestPriorModel = None
            if sprintIdx >= 1 :
                ( bestPriorModel , errScore ) = self . _hsState . bestModelInSprint ( 0 )
            if bestPriorModel is not None :
                self . logger . info ( "Best model and errScore from previous sprint(%d):" " %s, %g" % ( 0 , str ( bestPriorModel ) , errScore ) )
                ( baseState , modelId , errScore , completed , matured ) = self . _resultsDB . getParticleInfo ( bestPriorModel )
                particle . copyEncoderStatesFrom ( baseState )
                # Copy the best inference type from the earlier sprint
                particle . copyVarStatesFrom ( baseState , [ 'modelParams|inferenceType' ] )
                # It ' s best to jiggle the best settings from the prior sprint , so
                # compute a new position starting from that previous best
                # Only jiggle the vars we copied from the prior model
                whichVars = [ ]
                for varName in baseState [ 'varStates' ] :
                    if ':' in varName :
                        whichVars . append ( varName )
                particle . newPosition ( whichVars )
                self . logger . debug ( "Particle after incorporating encoder vars from best " "model in previous sprint: \n%s" % ( str ( particle ) ) )
            return ( False , particle , swarmId )
        # 2 . ) Look for a completed particle to evolve
        # Note that we use lastDescendent . We only want to evolve particles that
        # are at their most recent generation index .
        ( readyParticles , readyModelIds , readyErrScores , _ , _ ) = ( self . _resultsDB . getParticleInfos ( swarmId , genIdx = None , matured = True , lastDescendent = True ) )
        # If we have at least 1 ready particle to evolve . . .
        if len ( readyParticles ) > 0 :
            readyGenIdxs = [ x [ 'genIdx' ] for x in readyParticles ]
            sortedGenIdxs = sorted ( set ( readyGenIdxs ) )
            genIdx = sortedGenIdxs [ 0 ]
            # Now , genIdx has the generation of the particle we want to run ,
            # Get a particle from that generation and evolve it .
            useParticle = None
            for particle in readyParticles :
                if particle [ 'genIdx' ] == genIdx :
                    useParticle = particle
                    break
            # If speculativeParticles is off , we don ' t want to evolve a particle
            # into the next generation until all particles in the current
            # generation have completed .
            if not self . _speculativeParticles :
                ( particles , _ , _ , _ , _ ) = self . _resultsDB . getParticleInfos ( swarmId , genIdx = genIdx , matured = False )
                if len ( particles ) > 0 :
                    continue
            particle = Particle ( hsObj = self , resultsDB = self . _resultsDB , flattenedPermuteVars = self . _flattenedPermutations , evolveFromState = useParticle )
            return ( False , particle , swarmId )
        # END : for ( swarmSize , swarmId ) in swarmSizeAndIdList :
        # No success in this swarm , onto next swarm
    # We couldn ' t find a particle in this sprint ready to evolve . If
    # speculative particles is OFF , we have to wait for one or more other
    # workers to finish up their particles before we can do anything .
    if not self . _speculativeParticles :
        self . logger . info ( "Waiting for one or more of the %s swarms " "to complete a generation before evolving any more particles" % ( str ( swarmIds ) ) )
        return ( False , None , None )


signature:
def _okToExit ( self ) :
    """Test if it ' s OK to exit this worker . This is only called when we run
 out of prospective new models to evaluate . This method sees if all models
 have matured yet . If not , it will sleep for a bit and return False . This
 will indicate to the hypersearch worker that we should keep running , and
 check again later . This gives this worker a chance to pick up and adopt any
 model which may become orphaned by another worker before it matures .
 If all models have matured , this method will send a STOP message to all
 matured , running models ( presummably , there will be just one - the model
 which thinks it ' s the best ) before returning True ."""

implementation:
# Send an update status periodically to the JobTracker so that it doesn ' t
# think this worker is dead .
print >> sys . stderr , "reporter:status:In hypersearchV2: _okToExit"
# Any immature models still running ?
if not self . _jobCancelled :
    ( _ , modelIds , _ , _ , _ ) = self . _resultsDB . getParticleInfos ( matured = False )
    if len ( modelIds ) > 0 :
        self . logger . info ( "Ready to end hyperseach, but not all models have " "matured yet. Sleeping a bit to wait for all models " "to mature." )
        # Sleep for a bit , no need to check for orphaned models very often
        time . sleep ( 5.0 * random . random ( ) )
        return False
# All particles have matured , send a STOP signal to any that are still
# running .
( _ , modelIds , _ , _ , _ ) = self . _resultsDB . getParticleInfos ( completed = False )
for modelId in modelIds :
    self . logger . info ( "Stopping model %d because the search has ended" % ( modelId ) )
    self . _cjDAO . modelSetFields ( modelId , dict ( engStop = ClientJobsDAO . STOP_REASON_STOPPED ) , ignoreUnchanged = True )
# Update the HsState to get the accurate field contributions .
self . _hsStatePeriodicUpdate ( )
pctFieldContributions , absFieldContributions = self . _hsState . getFieldContributions ( )
# Update the results field with the new field contributions .
jobResultsStr = self . _cjDAO . jobGetFields ( self . _jobID , [ 'results' ] ) [ 0 ]
if jobResultsStr is not None :
    jobResults = json . loads ( jobResultsStr )
else :
    jobResults = { }
# Update the fieldContributions field .
if pctFieldContributions != jobResults . get ( 'fieldContributions' , None ) :
    jobResults [ 'fieldContributions' ] = pctFieldContributions
    jobResults [ 'absoluteFieldContributions' ] = absFieldContributions
    isUpdated = self . _cjDAO . jobSetFieldIfEqual ( self . _jobID , fieldName = 'results' , curValue = jobResultsStr , newValue = json . dumps ( jobResults ) )
    if isUpdated :
        self . logger . info ( 'Successfully updated the field contributions:%s' , pctFieldContributions )
    else :
        self . logger . info ( 'Failed updating the field contributions, ' 'another hypersearch worker must have updated it' )
return True


signature:
def createModels ( self , numModels = 1 ) :
    """Create one or more new models for evaluation . These should NOT be models
 that we already know are in progress ( i . e . those that have been sent to us
 via recordModelProgress ) . We return a list of models to the caller
 ( HypersearchWorker ) and if one can be successfully inserted into
 the models table ( i . e . it is not a duplicate ) then HypersearchWorker will
 turn around and call our runModel ( ) method , passing in this model . If it
 is a duplicate , HypersearchWorker will call this method again . A model
 is a duplicate if either the modelParamsHash or particleHash is
 identical to another entry in the model table .
 The numModels is provided by HypersearchWorker as a suggestion as to how
 many models to generate . This particular implementation only ever returns 1
 model .
 Before choosing some new models , we first do a sweep for any models that
 may have been abandonded by failed workers . If / when we detect an abandoned
 model , we mark it as complete and orphaned and hide it from any subsequent
 queries to our ResultsDB . This effectively considers it as if it never
 existed . We also change the paramsHash and particleHash in the model record
 of the models table so that we can create another model with the same
 params and particle status and run it ( which we then do immediately ) .
 The modelParamsHash returned for each model should be a hash ( max allowed
 size of ClientJobsDAO . hashMaxSize ) that uniquely identifies this model by
 it ' s params and the optional particleHash should be a hash of the particleId
 and generation index . Every model that gets placed into the models database ,
 either by this worker or another worker , will have these hashes computed for
 it . The recordModelProgress gets called for every model in the database and
 the hash is used to tell which , if any , are the same as the ones this worker
 generated .
 NOTE : We check first ourselves for possible duplicates using the paramsHash
 before we return a model . If HypersearchWorker failed to insert it ( because
 some other worker beat us to it ) , it will turn around and call our
 recordModelProgress with that other model so that we now know about it . It
 will then call createModels ( ) again .
 This methods returns an exit boolean and the model to evaluate . If there is
 no model to evalulate , we may return False for exit because we want to stay
 alive for a while , waiting for all other models to finish . This gives us
 a chance to detect and pick up any possibly orphaned model by another
 worker .
 Parameters :
 numModels : number of models to generate
 retval : ( exit , models )
 exit : true if this worker should exit .
 models : list of tuples , one for each model . Each tuple contains :
 ( modelParams , modelParamsHash , particleHash )
 modelParams is a dictionary containing the following elements :
 structuredParams : dictionary containing all variables for
 this model , with encoders represented as a dict within
 this dict ( or None if they are not included .
 particleState : dictionary containing the state of this
 particle . This includes the position and velocity of
 each of it ' s variables , the particleId , and the particle
 generation index . It contains the following keys :
 id : The particle Id of the particle we are using to
 generate / track this model . This is a string of the
 form < hypesearchWorkerId > . < particleIdx >
 genIdx : the particle ' s generation index . This starts at 0
 and increments every time we move the particle to a
 new position .
 swarmId : The swarmId , which is a string of the form
 < encoder > . < encoder > . . . that describes this swarm
 varStates : dict of the variable states . The key is the
 variable name , the value is a dict of the variable ' s
 position , velocity , bestPosition , bestResult , etc ."""

implementation:
# Check for and mark orphaned models
self . _checkForOrphanedModels ( )
modelResults = [ ]
for _ in xrange ( numModels ) :
    candidateParticle = None
    # If we ' ve reached the max # of model to evaluate , we ' re done .
    if ( self . _maxModels is not None and ( self . _resultsDB . numModels ( ) - self . _resultsDB . getNumErrModels ( ) ) >= self . _maxModels ) :
        return ( self . _okToExit ( ) , [ ] )
    # If we don ' t already have a particle to work on , get a candidate swarm and
    # particle to work with . If None is returned for the particle it means
    # either that the search is over ( if exitNow is also True ) or that we need
    # to wait for other workers to finish up their models before we can pick
    # another particle to run ( if exitNow is False ) .
    if candidateParticle is None :
        ( exitNow , candidateParticle , candidateSwarm ) = ( self . _getCandidateParticleAndSwarm ( ) )
    if candidateParticle is None :
        if exitNow :
            return ( self . _okToExit ( ) , [ ] )
        else : # Send an update status periodically to the JobTracker so that it doesn ' t
        # think this worker is dead .
            print >> sys . stderr , "reporter:status:In hypersearchV2: speculativeWait"
            time . sleep ( self . _speculativeWaitSecondsMax * random . random ( ) )
            return ( False , [ ] )
    useEncoders = candidateSwarm . split ( '.' )
    numAttempts = 0
    # Loop until we can create a unique model that we haven ' t seen yet .
    while True : # If this is the Nth attempt with the same candidate , agitate it a bit
    # to find a new unique position for it .
        if numAttempts >= 1 :
            self . logger . debug ( "Agitating particle to get unique position after %d " "failed attempts in a row" % ( numAttempts ) )
            candidateParticle . agitate ( )
        # Create the hierarchical params expected by the base description . Note
        # that this is where we incorporate encoders that have no permuted
        # values in them .
        position = candidateParticle . getPosition ( )
        structuredParams = dict ( )
        def _buildStructuredParams ( value , keys ) :
            flatKey = _flattenKeys ( keys )
            # If it ' s an encoder , either put in None if it ' s not used , or replace
            # all permuted constructor params with the actual position .
            if flatKey in self . _encoderNames :
                if flatKey in useEncoders : # Form encoder dict , substituting in chosen permutation values .
                    return value . getDict ( flatKey , position )
                # Encoder not used .
                else :
                    return None
            # Regular top - level variable .
            elif flatKey in position :
                return position [ flatKey ]
            # Fixed override of a parameter in the base description .
            else :
                return value
        structuredParams = rCopy ( self . _permutations , _buildStructuredParams , discardNoneKeys = False )
        # Create the modelParams .
        modelParams = dict ( structuredParams = structuredParams , particleState = candidateParticle . getState ( ) )
        # And the hashes .
        m = hashlib . md5 ( )
        m . update ( sortedJSONDumpS ( structuredParams ) )
        m . update ( self . _baseDescriptionHash )
        paramsHash = m . digest ( )
        particleInst = "%s.%s" % ( modelParams [ 'particleState' ] [ 'id' ] , modelParams [ 'particleState' ] [ 'genIdx' ] )
        particleHash = hashlib . md5 ( particleInst ) . digest ( )
        # Increase attempt counter
        numAttempts += 1
        # If this is a new one , and passes the filter test , exit with it .
        # TODO : There is currently a problem with this filters implementation as
        # it relates to self . _ maxUniqueModelAttempts . When there is a filter in
        # effect , we should try a lot more times before we decide we have
        # exhausted the parameter space for this swarm . The question is , how many
        # more times ?
        if self . _filterFunc and not self . _filterFunc ( structuredParams ) :
            valid = False
        else :
            valid = True
        if valid and self . _resultsDB . getModelIDFromParamsHash ( paramsHash ) is None :
            break
        # If we ' ve exceeded the max allowed number of attempts , mark this swarm
        # as completing or completed , so we don ' t try and allocate any more new
        # particles to it , and pick another .
        if numAttempts >= self . _maxUniqueModelAttempts :
            ( exitNow , candidateParticle , candidateSwarm ) = self . _getCandidateParticleAndSwarm ( exhaustedSwarmId = candidateSwarm )
            if candidateParticle is None :
                if exitNow :
                    return ( self . _okToExit ( ) , [ ] )
                else :
                    time . sleep ( self . _speculativeWaitSecondsMax * random . random ( ) )
                    return ( False , [ ] )
            numAttempts = 0
            useEncoders = candidateSwarm . split ( '.' )
    # Log message
    if self . logger . getEffectiveLevel ( ) <= logging . DEBUG :
        self . logger . debug ( "Submitting new potential model to HypersearchWorker: \n%s" % ( pprint . pformat ( modelParams , indent = 4 ) ) )
    modelResults . append ( ( modelParams , paramsHash , particleHash ) )
return ( False , modelResults )


signature:
def recordModelProgress ( self , modelID , modelParams , modelParamsHash , results , completed , completionReason , matured , numRecords ) :
    """Record or update the results for a model . This is called by the
 HSW whenever it gets results info for another model , or updated results
 on a model that is still running .
 The first time this is called for a given modelID , the modelParams will
 contain the params dict for that model and the modelParamsHash will contain
 the hash of the params . Subsequent updates of the same modelID will
 have params and paramsHash values of None ( in order to save overhead ) .
 The Hypersearch object should save these results into it ' s own working
 memory into some table , which it then uses to determine what kind of
 new models to create next time createModels ( ) is called .
 Parameters :
 modelID : ID of this model in models table
 modelParams : params dict for this model , or None if this is just an update
 of a model that it already previously reported on .
 See the comments for the createModels ( ) method for a
 description of this dict .
 modelParamsHash : hash of the modelParams dict , generated by the worker
 that put it into the model database .
 results : tuple containing ( allMetrics , optimizeMetric ) . Each is a
 dict containing metricName : result pairs . .
 May be none if we have no results yet .
 completed : True if the model has completed evaluation , False if it
 is still running ( and these are online results )
 completionReason : One of the ClientJobsDAO . CMPL _ REASON _ XXX equates
 matured : True if this model has matured . In most cases , once a
 model matures , it will complete as well . The only time a
 model matures and does not complete is if it ' s currently
 the best model and we choose to keep it running to generate
 predictions .
 numRecords : Number of records that have been processed so far by this
 model ."""

implementation:
if results is None :
    metricResult = None
else :
    metricResult = results [ 1 ] . values ( ) [ 0 ]
# Update our database .
errScore = self . _resultsDB . update ( modelID = modelID , modelParams = modelParams , modelParamsHash = modelParamsHash , metricResult = metricResult , completed = completed , completionReason = completionReason , matured = matured , numRecords = numRecords )
# Log message .
self . logger . debug ( 'Received progress on model %d: completed: %s, ' 'cmpReason: %s, numRecords: %d, errScore: %s' , modelID , completed , completionReason , numRecords , errScore )
# Log best so far .
( bestModelID , bestResult ) = self . _resultsDB . bestModelIdAndErrScore ( )
self . logger . debug ( 'Best err score seen so far: %s on model %s' % ( bestResult , bestModelID ) )


signature:
def runModel ( self , modelID , jobID , modelParams , modelParamsHash , jobsDAO , modelCheckpointGUID ) :
    """Run the given model .
 This runs the model described by ' modelParams ' . Periodically , it updates
 the results seen on the model to the model database using the databaseAO
 ( database Access Object ) methods .
 Parameters :
 modelID : ID of this model in models table
 jobID : ID for this hypersearch job in the jobs table
 modelParams : parameters of this specific model
 modelParams is a dictionary containing the name / value
 pairs of each variable we are permuting over . Note that
 variables within an encoder spec have their name
 structure as :
 < encoderName > . < encodrVarName >
 modelParamsHash : hash of modelParamValues
 jobsDAO jobs data access object - the interface to the jobs
 database where model information is stored
 modelCheckpointGUID : A persistent , globally - unique identifier for
 constructing the model checkpoint key"""

implementation:
# We ' re going to make an assumption that if we ' re not using streams , that
# we also don ' t need checkpoints saved . For now , this assumption is OK
# ( if there are no streams , we ' re typically running on a single machine
# and just save models to files ) but we may want to break this out as
# a separate controllable parameter in the future
if not self . _createCheckpoints :
    modelCheckpointGUID = None
# Register this model in our database
self . _resultsDB . update ( modelID = modelID , modelParams = modelParams , modelParamsHash = modelParamsHash , metricResult = None , completed = False , completionReason = None , matured = False , numRecords = 0 )
# Get the structured params , which we pass to the base description
structuredParams = modelParams [ 'structuredParams' ]
if self . logger . getEffectiveLevel ( ) <= logging . DEBUG :
    self . logger . debug ( "Running Model. \nmodelParams: %s, \nmodelID=%s, " % ( pprint . pformat ( modelParams , indent = 4 ) , modelID ) )
# Record time . clock ( ) so that we can report on cpu time
cpuTimeStart = time . clock ( )
# Run the experiment . This will report the results back to the models
# database for us as well .
logLevel = self . logger . getEffectiveLevel ( )
try :
    if self . _dummyModel is None or self . _dummyModel is False :
        ( cmpReason , cmpMsg ) = runModelGivenBaseAndParams ( modelID = modelID , jobID = jobID , baseDescription = self . _baseDescription , params = structuredParams , predictedField = self . _predictedField , reportKeys = self . _reportKeys , optimizeKey = self . _optimizeKey , jobsDAO = jobsDAO , modelCheckpointGUID = modelCheckpointGUID , logLevel = logLevel , predictionCacheMaxRecords = self . _predictionCacheMaxRecords )
    else :
        dummyParams = dict ( self . _dummyModel )
        dummyParams [ 'permutationParams' ] = structuredParams
        if self . _dummyModelParamsFunc is not None :
            permInfo = dict ( structuredParams )
            permInfo [ 'generation' ] = modelParams [ 'particleState' ] [ 'genIdx' ]
            dummyParams . update ( self . _dummyModelParamsFunc ( permInfo ) )
        ( cmpReason , cmpMsg ) = runDummyModel ( modelID = modelID , jobID = jobID , params = dummyParams , predictedField = self . _predictedField , reportKeys = self . _reportKeys , optimizeKey = self . _optimizeKey , jobsDAO = jobsDAO , modelCheckpointGUID = modelCheckpointGUID , logLevel = logLevel , predictionCacheMaxRecords = self . _predictionCacheMaxRecords )
    # Write out the completion reason and message
    jobsDAO . modelSetCompleted ( modelID , completionReason = cmpReason , completionMsg = cmpMsg , cpuTime = time . clock ( ) - cpuTimeStart )
except InvalidConnectionException , e :
    self . logger . warn ( "%s" , e )


signature:
def _escape ( s ) :
    """Escape commas , tabs , newlines and dashes in a string
 Commas are encoded as tabs"""

implementation:
assert isinstance ( s , str ) , "expected %s but got %s; value=%s" % ( type ( str ) , type ( s ) , s )
s = s . replace ( "\\" , "\\\\" )
s = s . replace ( "\n" , "\\n" )
s = s . replace ( "\t" , "\\t" )
s = s . replace ( "," , "\t" )
return s


signature:
def _engineServicesRunning ( ) :
    """Return true if the engine services are running"""

implementation:
process = subprocess . Popen ( [ "ps" , "aux" ] , stdout = subprocess . PIPE )
stdout = process . communicate ( ) [ 0 ]
result = process . returncode
if result != 0 :
    raise RuntimeError ( "Unable to check for running client job manager" )
# See if the CJM is running
running = False
for line in stdout . split ( "\n" ) :
    if "python" in line and "clientjobmanager.client_job_manager" in line :
        running = True
        break
return running


signature:
def runWithConfig ( swarmConfig , options , outDir = None , outputLabel = "default" , permWorkDir = None , verbosity = 1 ) :
    """Starts a swarm , given an dictionary configuration .
 @ param swarmConfig { dict } A complete [ swarm description ] ( http : / / nupic . docs . numenta . org / 0.7.0 . dev0 / guides / swarming / running . html # the - swarm - description ) object .
 @ param outDir { string } Optional path to write swarm details ( defaults to
 current working directory ) .
 @ param outputLabel { string } Optional label for output ( defaults to " default " ) .
 @ param permWorkDir { string } Optional location of working directory ( defaults
 to current working directory ) .
 @ param verbosity { int } Optional ( 1,2,3 ) increasing verbosity of output .
 @ returns { object } Model parameters"""

implementation:
global g_currentVerbosityLevel
g_currentVerbosityLevel = verbosity
# Generate the description and permutations . py files in the same directory
# for reference .
if outDir is None :
    outDir = os . getcwd ( )
if permWorkDir is None :
    permWorkDir = os . getcwd ( )
_checkOverwrite ( options , outDir )
_generateExpFilesFromSwarmDescription ( swarmConfig , outDir )
options [ "expDescConfig" ] = swarmConfig
options [ "outputLabel" ] = outputLabel
options [ "outDir" ] = outDir
options [ "permWorkDir" ] = permWorkDir
runOptions = _injectDefaultOptions ( options )
_validateOptions ( runOptions )
return _runAction ( runOptions )


signature:
def runWithJsonFile ( expJsonFilePath , options , outputLabel , permWorkDir ) :
    """Starts a swarm , given a path to a JSON file containing configuration .
 This function is meant to be used with a CLI wrapper that passes command line
 arguments in through the options parameter .
 @ param expJsonFilePath { string } Path to a JSON file containing the complete
 [ swarm description ] ( http : / / nupic . docs . numenta . org / 0.7.0 . dev0 / guides / swarming / running . html # the - swarm - description ) .
 @ param options { dict } CLI options .
 @ param outputLabel { string } Label for output .
 @ param permWorkDir { string } Location of working directory .
 @ returns { int } Swarm job id ."""

implementation:
if "verbosityCount" in options :
    verbosity = options [ "verbosityCount" ]
    del options [ "verbosityCount" ]
else :
    verbosity = 1
_setupInterruptHandling ( )
with open ( expJsonFilePath , "r" ) as jsonFile :
    expJsonConfig = json . loads ( jsonFile . read ( ) )
outDir = os . path . dirname ( expJsonFilePath )
return runWithConfig ( expJsonConfig , options , outDir = outDir , outputLabel = outputLabel , permWorkDir = permWorkDir , verbosity = verbosity )


signature:
def runWithPermutationsScript ( permutationsFilePath , options , outputLabel , permWorkDir ) :
    """Starts a swarm , given a path to a permutations . py script .
 This function is meant to be used with a CLI wrapper that passes command line
 arguments in through the options parameter .
 @ param permutationsFilePath { string } Path to permutations . py .
 @ param options { dict } CLI options .
 @ param outputLabel { string } Label for output .
 @ param permWorkDir { string } Location of working directory .
 @ returns { object } Model parameters ."""

implementation:
global g_currentVerbosityLevel
if "verbosityCount" in options :
    g_currentVerbosityLevel = options [ "verbosityCount" ]
    del options [ "verbosityCount" ]
else :
    g_currentVerbosityLevel = 1
_setupInterruptHandling ( )
options [ "permutationsScriptPath" ] = permutationsFilePath
options [ "outputLabel" ] = outputLabel
options [ "outDir" ] = permWorkDir
options [ "permWorkDir" ] = permWorkDir
# Assume it ' s a permutations python script
runOptions = _injectDefaultOptions ( options )
_validateOptions ( runOptions )
return _runAction ( runOptions )


signature:
def _backupFile ( filePath ) :
    """Back up a file
 Parameters :
 retval : Filepath of the back - up"""

implementation:
assert os . path . exists ( filePath )
stampNum = 0
( prefix , suffix ) = os . path . splitext ( filePath )
while True :
    backupPath = "%s.%d%s" % ( prefix , stampNum , suffix )
    stampNum += 1
    if not os . path . exists ( backupPath ) :
        break
shutil . copyfile ( filePath , backupPath )
return backupPath


signature:
def pickupSearch ( self ) :
    """Pick up the latest search from a saved jobID and monitor it to completion
 Parameters :
 retval : nothing"""

implementation:
self . __searchJob = self . loadSavedHyperSearchJob ( permWorkDir = self . _options [ "permWorkDir" ] , outputLabel = self . _options [ "outputLabel" ] )
self . monitorSearchJob ( )


signature:
def monitorSearchJob ( self ) :
    """Parameters :
 retval : nothing"""

implementation:
assert self . __searchJob is not None
jobID = self . __searchJob . getJobID ( )
startTime = time . time ( )
lastUpdateTime = datetime . now ( )
# Monitor HyperSearch and report progress
# NOTE : may be - 1 if it can ' t be determined
expectedNumModels = self . __searchJob . getExpectedNumModels ( searchMethod = self . _options [ "searchMethod" ] )
lastNumFinished = 0
finishedModelIDs = set ( )
finishedModelStats = _ModelStats ( )
# Keep track of the worker state , results , and milestones from the job
# record
lastWorkerState = None
lastJobResults = None
lastModelMilestones = None
lastEngStatus = None
hyperSearchFinished = False
while not hyperSearchFinished :
    jobInfo = self . __searchJob . getJobStatus ( self . _workers )
    # Check for job completion BEFORE processing models ; NOTE : this permits us
    # to process any models that we may not have accounted for in the
    # previous iteration .
    hyperSearchFinished = jobInfo . isFinished ( )
    # Look for newly completed models , and process them
    modelIDs = self . __searchJob . queryModelIDs ( )
    _emit ( Verbosity . DEBUG , "Current number of models is %d (%d of them completed)" % ( len ( modelIDs ) , len ( finishedModelIDs ) ) )
    if len ( modelIDs ) > 0 : # Build a list of modelIDs to check for completion
        checkModelIDs = [ ]
        for modelID in modelIDs :
            if modelID not in finishedModelIDs :
                checkModelIDs . append ( modelID )
        del modelIDs
        # Process newly completed models
        if checkModelIDs :
            _emit ( Verbosity . DEBUG , "Checking %d models..." % ( len ( checkModelIDs ) ) )
            errorCompletionMsg = None
            for ( i , modelInfo ) in enumerate ( _iterModels ( checkModelIDs ) ) :
                _emit ( Verbosity . DEBUG , "[%s] Checking completion: %s" % ( i , modelInfo ) )
                if modelInfo . isFinished ( ) :
                    finishedModelIDs . add ( modelInfo . getModelID ( ) )
                    finishedModelStats . update ( modelInfo )
                    if ( modelInfo . getCompletionReason ( ) . isError ( ) and not errorCompletionMsg ) :
                        errorCompletionMsg = modelInfo . getCompletionMsg ( )
                    # Update the set of all encountered metrics keys ( we will use
                    # these to print column names in reports . csv )
                    metrics = modelInfo . getReportMetrics ( )
                    self . __foundMetrcsKeySet . update ( metrics . keys ( ) )
        numFinished = len ( finishedModelIDs )
        # Print current completion stats
        if numFinished != lastNumFinished :
            lastNumFinished = numFinished
            if expectedNumModels is None :
                expModelsStr = ""
            else :
                expModelsStr = "of %s" % ( expectedNumModels )
            stats = finishedModelStats
            print ( "<jobID: %s> %s %s models finished [success: %s; %s: %s; %s: " "%s; %s: %s; %s: %s; %s: %s; %s: %s]" % ( jobID , numFinished , expModelsStr , # stats . numCompletedSuccess ,
            ( stats . numCompletedEOF + stats . numCompletedStopped ) , "EOF" if stats . numCompletedEOF else "eof" , stats . numCompletedEOF , "STOPPED" if stats . numCompletedStopped else "stopped" , stats . numCompletedStopped , "KILLED" if stats . numCompletedKilled else "killed" , stats . numCompletedKilled , "ERROR" if stats . numCompletedError else "error" , stats . numCompletedError , "ORPHANED" if stats . numCompletedError else "orphaned" , stats . numCompletedOrphaned , "UNKNOWN" if stats . numCompletedOther else "unknown" , stats . numCompletedOther ) )
            # Print the first error message from the latest batch of completed
            # models
            if errorCompletionMsg :
                print "ERROR MESSAGE: %s" % errorCompletionMsg
        # Print the new worker state , if it changed
        workerState = jobInfo . getWorkerState ( )
        if workerState != lastWorkerState :
            print "##>> UPDATED WORKER STATE: \n%s" % ( pprint . pformat ( workerState , indent = 4 ) )
            lastWorkerState = workerState
        # Print the new job results , if it changed
        jobResults = jobInfo . getResults ( )
        if jobResults != lastJobResults :
            print "####>> UPDATED JOB RESULTS: \n%s (elapsed time: %g secs)" % ( pprint . pformat ( jobResults , indent = 4 ) , time . time ( ) - startTime )
            lastJobResults = jobResults
        # Print the new model milestones if they changed
        modelMilestones = jobInfo . getModelMilestones ( )
        if modelMilestones != lastModelMilestones :
            print "##>> UPDATED MODEL MILESTONES: \n%s" % ( pprint . pformat ( modelMilestones , indent = 4 ) )
            lastModelMilestones = modelMilestones
        # Print the new engine status if it changed
        engStatus = jobInfo . getEngStatus ( )
        if engStatus != lastEngStatus :
            print "##>> UPDATED STATUS: \n%s" % ( engStatus )
            lastEngStatus = engStatus
    # Sleep before next check
    if not hyperSearchFinished :
        if self . _options [ "timeout" ] != None :
            if ( ( datetime . now ( ) - lastUpdateTime ) > timedelta ( minutes = self . _options [ "timeout" ] ) ) :
                print "Timeout reached, exiting"
                self . __cjDAO . jobCancel ( jobID )
                sys . exit ( 1 )
        time . sleep ( 1 )
# Tabulate results
modelIDs = self . __searchJob . queryModelIDs ( )
print "Evaluated %s models" % len ( modelIDs )
print "HyperSearch finished!"
jobInfo = self . __searchJob . getJobStatus ( self . _workers )
print "Worker completion message: %s" % ( jobInfo . getWorkerCompletionMsg ( ) )


signature:
def _launchWorkers ( self , cmdLine , numWorkers ) :
    """Launch worker processes to execute the given command line
 Parameters :
 cmdLine : The command line for each worker
 numWorkers : number of workers to launch"""

implementation:
self . _workers = [ ]
for i in range ( numWorkers ) :
    stdout = tempfile . NamedTemporaryFile ( delete = False )
    stderr = tempfile . NamedTemporaryFile ( delete = False )
    p = subprocess . Popen ( cmdLine , bufsize = 1 , env = os . environ , shell = True , stdin = None , stdout = stdout , stderr = stderr )
    p . _stderr_file = stderr
    p . _stdout_file = stdout
    self . _workers . append ( p )


signature:
def __startSearch ( self ) :
    """Starts HyperSearch as a worker or runs it inline for the " dryRun " action
 Parameters :
 retval : the new _ HyperSearchJob instance representing the
 HyperSearch job"""

implementation:
# This search uses a pre - existing permutations script
params = _ClientJobUtils . makeSearchJobParamsDict ( options = self . _options , forRunning = True )
if self . _options [ "action" ] == "dryRun" :
    args = [ sys . argv [ 0 ] , "--params=%s" % ( json . dumps ( params ) ) ]
    print
    print "=================================================================="
    print "RUNNING PERMUTATIONS INLINE as \"DRY RUN\"..."
    print "=================================================================="
    jobID = hypersearch_worker . main ( args )
else :
    cmdLine = _setUpExports ( self . _options [ "exports" ] )
    # Begin the new search . The { JOBID } string is replaced by the actual
    # jobID returned from jobInsert .
    cmdLine += "$HYPERSEARCH"
    maxWorkers = self . _options [ "maxWorkers" ]
    jobID = self . __cjDAO . jobInsert ( client = "GRP" , cmdLine = cmdLine , params = json . dumps ( params ) , minimumWorkers = 1 , maximumWorkers = maxWorkers , jobType = self . __cjDAO . JOB_TYPE_HS )
    cmdLine = "python -m nupic.swarming.hypersearch_worker" " --jobID=%d" % ( jobID )
    self . _launchWorkers ( cmdLine , maxWorkers )
searchJob = _HyperSearchJob ( jobID )
# Save search ID to file ( this is used for report generation )
self . __saveHyperSearchJobID ( permWorkDir = self . _options [ "permWorkDir" ] , outputLabel = self . _options [ "outputLabel" ] , hyperSearchJob = searchJob )
if self . _options [ "action" ] == "dryRun" :
    print "Successfully executed \"dry-run\" hypersearch, jobID=%d" % ( jobID )
else :
    print "Successfully submitted new HyperSearch job, jobID=%d" % ( jobID )
    _emit ( Verbosity . DEBUG , "Each worker executing the command line: %s" % ( cmdLine , ) )
return searchJob


signature:
def generateReport ( cls , options , replaceReport , hyperSearchJob , metricsKeys ) :
    """Prints all available results in the given HyperSearch job and emits
 model information to the permutations report csv .
 The job may be completed or still in progress .
 Parameters :
 options : NupicRunPermutations options dict
 replaceReport : True to replace existing report csv , if any ; False to
 append to existing report csv , if any
 hyperSearchJob : _ HyperSearchJob instance ; if None , will get it from saved
 jobID , if any
 metricsKeys : sequence of report metrics key names to include in report ;
 if None , will pre - scan all modelInfos to generate a complete
 list of metrics key names .
 retval : model parameters"""

implementation:
# Load _ HyperSearchJob instance from storage , if not provided
if hyperSearchJob is None :
    hyperSearchJob = cls . loadSavedHyperSearchJob ( permWorkDir = options [ "permWorkDir" ] , outputLabel = options [ "outputLabel" ] )
modelIDs = hyperSearchJob . queryModelIDs ( )
bestModel = None
# If metricsKeys was not provided , pre - scan modelInfos to create the list ;
# this is needed by _ ReportCSVWriter
# Also scan the parameters to generate a list of encoders and search
# parameters
metricstmp = set ( )
searchVar = set ( )
for modelInfo in _iterModels ( modelIDs ) :
    if modelInfo . isFinished ( ) :
        vars = modelInfo . getParamLabels ( ) . keys ( )
        searchVar . update ( vars )
        metrics = modelInfo . getReportMetrics ( )
        metricstmp . update ( metrics . keys ( ) )
if metricsKeys is None :
    metricsKeys = metricstmp
# Create a csv report writer
reportWriter = _ReportCSVWriter ( hyperSearchJob = hyperSearchJob , metricsKeys = metricsKeys , searchVar = searchVar , outputDirAbsPath = options [ "permWorkDir" ] , outputLabel = options [ "outputLabel" ] , replaceReport = replaceReport )
# Tallies of experiment dispositions
modelStats = _ModelStats ( )
# numCompletedOther = long ( 0)
print "\nResults from all experiments:"
print "----------------------------------------------------------------"
# Get common optimization metric info from permutations script
searchParams = hyperSearchJob . getParams ( )
( optimizationMetricKey , maximizeMetric ) = ( _PermutationUtils . getOptimizationMetricInfo ( searchParams ) )
# Print metrics , while looking for the best model
formatStr = None
# NOTE : we may find additional metrics if HyperSearch is still running
foundMetricsKeySet = set ( metricsKeys )
sortedMetricsKeys = [ ]
# pull out best Model from jobs table
jobInfo = _clientJobsDB ( ) . jobInfo ( hyperSearchJob . getJobID ( ) )
# Try to return a decent error message if the job was cancelled for some
# reason .
if jobInfo . cancel == 1 :
    raise Exception ( jobInfo . workerCompletionMsg )
try :
    results = json . loads ( jobInfo . results )
except Exception , e :
    print "json.loads(jobInfo.results) raised an exception.  " "Here is some info to help with debugging:"
    print "jobInfo: " , jobInfo
    print "jobInfo.results: " , jobInfo . results
    print "EXCEPTION: " , e
    raise
bestModelNum = results [ "bestModel" ]
bestModelIterIndex = None
# performance metrics for the entire job
totalWallTime = 0
totalRecords = 0
# At the end , we will sort the models by their score on the optimization
# metric
scoreModelIDDescList = [ ]
for ( i , modelInfo ) in enumerate ( _iterModels ( modelIDs ) ) : # Output model info to report csv
    reportWriter . emit ( modelInfo )
    # Update job metrics
    totalRecords += modelInfo . getNumRecords ( )
    format = "%Y-%m-%d %H:%M:%S"
    startTime = modelInfo . getStartTime ( )
    if modelInfo . isFinished ( ) :
        endTime = modelInfo . getEndTime ( )
        st = datetime . strptime ( startTime , format )
        et = datetime . strptime ( endTime , format )
        totalWallTime += ( et - st ) . seconds
    # Tabulate experiment dispositions
    modelStats . update ( modelInfo )
    # For convenience
    expDesc = modelInfo . getModelDescription ( )
    reportMetrics = modelInfo . getReportMetrics ( )
    optimizationMetrics = modelInfo . getOptimizationMetrics ( )
    if modelInfo . getModelID ( ) == bestModelNum :
        bestModel = modelInfo
        bestModelIterIndex = i
        bestMetric = optimizationMetrics . values ( ) [ 0 ]
    # Keep track of the best - performing model
    if optimizationMetrics :
        assert len ( optimizationMetrics ) == 1 , ( "expected 1 opt key, but got %d (%s) in %s" % ( len ( optimizationMetrics ) , optimizationMetrics , modelInfo ) )
    # Append to our list of modelIDs and scores
    if modelInfo . getCompletionReason ( ) . isEOF ( ) :
        scoreModelIDDescList . append ( ( optimizationMetrics . values ( ) [ 0 ] , modelInfo . getModelID ( ) , modelInfo . getGeneratedDescriptionFile ( ) , modelInfo . getParamLabels ( ) ) )
    print "[%d] Experiment %s\n(%s):" % ( i , modelInfo , expDesc )
    if ( modelInfo . isFinished ( ) and not ( modelInfo . getCompletionReason ( ) . isStopped or modelInfo . getCompletionReason ( ) . isEOF ( ) ) ) :
        print ">> COMPLETION MESSAGE: %s" % modelInfo . getCompletionMsg ( )
    if reportMetrics : # Update our metrics key set and format string
        foundMetricsKeySet . update ( reportMetrics . iterkeys ( ) )
        if len ( sortedMetricsKeys ) != len ( foundMetricsKeySet ) :
            sortedMetricsKeys = sorted ( foundMetricsKeySet )
            maxKeyLen = max ( [ len ( k ) for k in sortedMetricsKeys ] )
            formatStr = "  %%-%ds" % ( maxKeyLen + 2 )
        # Print metrics
        for key in sortedMetricsKeys :
            if key in reportMetrics :
                if key == optimizationMetricKey :
                    m = "%r (*)" % reportMetrics [ key ]
                else :
                    m = "%r" % reportMetrics [ key ]
                print formatStr % ( key + ":" ) , m
        print
# Summarize results
print "--------------------------------------------------------------"
if len ( modelIDs ) > 0 :
    print "%d experiments total (%s).\n" % ( len ( modelIDs ) , ( "all completed successfully" if ( modelStats . numCompletedKilled + modelStats . numCompletedEOF ) == len ( modelIDs ) else "WARNING: %d models have not completed or there were errors" % ( len ( modelIDs ) - ( modelStats . numCompletedKilled + modelStats . numCompletedEOF + modelStats . numCompletedStopped ) ) ) )
    if modelStats . numStatusOther > 0 :
        print "ERROR: models with unexpected status: %d" % ( modelStats . numStatusOther )
    print "WaitingToStart: %d" % modelStats . numStatusWaitingToStart
    print "Running: %d" % modelStats . numStatusRunning
    print "Completed: %d" % modelStats . numStatusCompleted
    if modelStats . numCompletedOther > 0 :
        print "    ERROR: models with unexpected completion reason: %d" % ( modelStats . numCompletedOther )
    print "    ran to EOF: %d" % modelStats . numCompletedEOF
    print "    ran to stop signal: %d" % modelStats . numCompletedStopped
    print "    were orphaned: %d" % modelStats . numCompletedOrphaned
    print "    killed off: %d" % modelStats . numCompletedKilled
    print "    failed: %d" % modelStats . numCompletedError
    assert modelStats . numStatusOther == 0 , "numStatusOther=%s" % ( modelStats . numStatusOther )
    assert modelStats . numCompletedOther == 0 , "numCompletedOther=%s" % ( modelStats . numCompletedOther )
else :
    print "0 experiments total."
# Print out the field contributions
print
global gCurrentSearch
jobStatus = hyperSearchJob . getJobStatus ( gCurrentSearch . _workers )
jobResults = jobStatus . getResults ( )
if "fieldContributions" in jobResults :
    print "Field Contributions:"
    pprint . pprint ( jobResults [ "fieldContributions" ] , indent = 4 )
else :
    print "Field contributions info not available"
# Did we have an optimize key ?
if bestModel is not None :
    maxKeyLen = max ( [ len ( k ) for k in sortedMetricsKeys ] )
    maxKeyLen = max ( maxKeyLen , len ( optimizationMetricKey ) )
    formatStr = "  %%-%ds" % ( maxKeyLen + 2 )
    bestMetricValue = bestModel . getOptimizationMetrics ( ) . values ( ) [ 0 ]
    optimizationMetricName = bestModel . getOptimizationMetrics ( ) . keys ( ) [ 0 ]
    print
    print "Best results on the optimization metric %s (maximize=%s):" % ( optimizationMetricName , maximizeMetric )
    print "[%d] Experiment %s (%s):" % ( bestModelIterIndex , bestModel , bestModel . getModelDescription ( ) )
    print formatStr % ( optimizationMetricName + ":" ) , bestMetricValue
    print
    print "Total number of Records processed: %d" % totalRecords
    print
    print "Total wall time for all models: %d" % totalWallTime
    hsJobParams = hyperSearchJob . getParams ( )
# Were we asked to write out the top N model description files ?
if options [ "genTopNDescriptions" ] > 0 :
    print "\nGenerating description files for top %d models..." % ( options [ "genTopNDescriptions" ] )
    scoreModelIDDescList . sort ( )
    scoreModelIDDescList = scoreModelIDDescList [ 0 : options [ "genTopNDescriptions" ] ]
    i = - 1
    for ( score , modelID , description , paramLabels ) in scoreModelIDDescList :
        i += 1
        outDir = os . path . join ( options [ "permWorkDir" ] , "model_%d" % ( i ) )
        print "Generating description file for model %s at %s" % ( modelID , outDir )
        if not os . path . exists ( outDir ) :
            os . makedirs ( outDir )
        # Fix up the location to the base description file .
        # importBaseDescription ( ) chooses the file relative to the calling file .
        # The calling file is in outDir .
        # The base description is in the user - specified " outDir "
        base_description_path = os . path . join ( options [ "outDir" ] , "description.py" )
        base_description_relpath = os . path . relpath ( base_description_path , start = outDir )
        description = description . replace ( "importBaseDescription('base.py', config)" , "importBaseDescription('%s', config)" % base_description_relpath )
        fd = open ( os . path . join ( outDir , "description.py" ) , "wb" )
        fd . write ( description )
        fd . close ( )
        # Generate a csv file with the parameter settings in it
        fd = open ( os . path . join ( outDir , "params.csv" ) , "wb" )
        writer = csv . writer ( fd )
        colNames = paramLabels . keys ( )
        colNames . sort ( )
        writer . writerow ( colNames )
        row = [ paramLabels [ x ] for x in colNames ]
        writer . writerow ( row )
        fd . close ( )
        print "Generating model params file..."
        # Generate a model params file alongside the description . py
        mod = imp . load_source ( "description" , os . path . join ( outDir , "description.py" ) )
        model_description = mod . descriptionInterface . getModelDescription ( )
        fd = open ( os . path . join ( outDir , "model_params.py" ) , "wb" )
        fd . write ( "%s\nMODEL_PARAMS = %s" % ( getCopyrightHead ( ) , pprint . pformat ( model_description ) ) )
        fd . close ( )
    print
reportWriter . finalize ( )
return model_description


signature:
def loadSavedHyperSearchJob ( cls , permWorkDir , outputLabel ) :
    """Instantiates a _ HyperSearchJob instance from info saved in file
 Parameters :
 permWorkDir : Directory path for saved jobID file
 outputLabel : Label string for incorporating into file name for saved jobID
 retval : _ HyperSearchJob instance ; raises exception if not found"""

implementation:
jobID = cls . __loadHyperSearchJobID ( permWorkDir = permWorkDir , outputLabel = outputLabel )
searchJob = _HyperSearchJob ( nupicJobID = jobID )
return searchJob


signature:
def __saveHyperSearchJobID ( cls , permWorkDir , outputLabel , hyperSearchJob ) :
    """Saves the given _ HyperSearchJob instance ' s jobID to file
 Parameters :
 permWorkDir : Directory path for saved jobID file
 outputLabel : Label string for incorporating into file name for saved jobID
 hyperSearchJob : _ HyperSearchJob instance
 retval : nothing"""

implementation:
jobID = hyperSearchJob . getJobID ( )
filePath = cls . __getHyperSearchJobIDFilePath ( permWorkDir = permWorkDir , outputLabel = outputLabel )
if os . path . exists ( filePath ) :
    _backupFile ( filePath )
d = dict ( hyperSearchJobID = jobID )
with open ( filePath , "wb" ) as jobIdPickleFile :
    pickle . dump ( d , jobIdPickleFile )


signature:
def __loadHyperSearchJobID ( cls , permWorkDir , outputLabel ) :
    """Loads a saved jobID from file
 Parameters :
 permWorkDir : Directory path for saved jobID file
 outputLabel : Label string for incorporating into file name for saved jobID
 retval : HyperSearch jobID ; raises exception if not found ."""

implementation:
filePath = cls . __getHyperSearchJobIDFilePath ( permWorkDir = permWorkDir , outputLabel = outputLabel )
jobID = None
with open ( filePath , "r" ) as jobIdPickleFile :
    jobInfo = pickle . load ( jobIdPickleFile )
    jobID = jobInfo [ "hyperSearchJobID" ]
return jobID


signature:
def __getHyperSearchJobIDFilePath ( cls , permWorkDir , outputLabel ) :
    """Returns filepath where to store HyperSearch JobID
 Parameters :
 permWorkDir : Directory path for saved jobID file
 outputLabel : Label string for incorporating into file name for saved jobID
 retval : Filepath where to store HyperSearch JobID"""

implementation:
# Get the base path and figure out the path of the report file .
basePath = permWorkDir
# Form the name of the output csv file that will contain all the results
filename = "%s_HyperSearchJobID.pkl" % ( outputLabel , )
filepath = os . path . join ( basePath , filename )
return filepath


signature:
def emit ( self , modelInfo ) :
    """Emit model info to csv file
 Parameters :
 modelInfo : _ NupicModelInfo instance
 retval : nothing"""

implementation:
# Open / init csv file , if needed
if self . __csvFileObj is None : # sets up self . _ _ sortedVariableNames and self . _ _ csvFileObj
    self . __openAndInitCSVFile ( modelInfo )
csv = self . __csvFileObj
# Emit model info row to report . csv
print >> csv , "%s, " % ( self . __searchJobID ) ,
print >> csv , "%s, " % ( modelInfo . getModelID ( ) ) ,
print >> csv , "%s, " % ( modelInfo . statusAsString ( ) ) ,
if modelInfo . isFinished ( ) :
    print >> csv , "%s, " % ( modelInfo . getCompletionReason ( ) ) ,
else :
    print >> csv , "NA, " ,
if not modelInfo . isWaitingToStart ( ) :
    print >> csv , "%s, " % ( modelInfo . getStartTime ( ) ) ,
else :
    print >> csv , "NA, " ,
if modelInfo . isFinished ( ) :
    dateFormat = "%Y-%m-%d %H:%M:%S"
    startTime = modelInfo . getStartTime ( )
    endTime = modelInfo . getEndTime ( )
    print >> csv , "%s, " % endTime ,
    st = datetime . strptime ( startTime , dateFormat )
    et = datetime . strptime ( endTime , dateFormat )
    print >> csv , "%s, " % ( str ( ( et - st ) . seconds ) ) ,
else :
    print >> csv , "NA, " ,
    print >> csv , "NA, " ,
print >> csv , "%s, " % str ( modelInfo . getModelDescription ( ) ) ,
print >> csv , "%s, " % str ( modelInfo . getNumRecords ( ) ) ,
paramLabelsDict = modelInfo . getParamLabels ( )
for key in self . __sortedVariableNames : # Some values are complex structures , . . which need to be represented as
# strings
    if key in paramLabelsDict :
        print >> csv , "%s, " % ( paramLabelsDict [ key ] ) ,
    else :
        print >> csv , "None, " ,
metrics = modelInfo . getReportMetrics ( )
for key in self . __sortedMetricsKeys :
    value = metrics . get ( key , "NA" )
    value = str ( value )
    value = value . replace ( "\n" , " " )
    print >> csv , "%s, " % ( value ) ,
print >> csv


signature:
def finalize ( self ) :
    """Close file and print report / backup csv file paths
 Parameters :
 retval : nothing"""

implementation:
if self . __csvFileObj is not None : # Done with file
    self . __csvFileObj . close ( )
    self . __csvFileObj = None
    print "Report csv saved in %s" % ( self . __reportCSVPath , )
    if self . __backupCSVPath :
        print "Previous report csv file was backed up to %s" % ( self . __backupCSVPath , )
else :
    print "Nothing was written to report csv file."


signature:
def __openAndInitCSVFile ( self , modelInfo ) :
    """- Backs up old report csv file ;
 - opens the report csv file in append or overwrite mode ( per
 self . _ _ replaceReport ) ;
 - emits column fields ;
 - sets up self . _ _ sortedVariableNames , self . _ _ csvFileObj ,
 self . _ _ backupCSVPath , and self . _ _ reportCSVPath
 Parameters :
 modelInfo : First _ NupicModelInfo instance passed to emit ( )
 retval : nothing"""

implementation:
# Get the base path and figure out the path of the report file .
basePath = self . __outputDirAbsPath
# Form the name of the output csv file that will contain all the results
reportCSVName = "%s_Report.csv" % ( self . __outputLabel , )
reportCSVPath = self . __reportCSVPath = os . path . join ( basePath , reportCSVName )
# If a report CSV file already exists , back it up
backupCSVPath = None
if os . path . exists ( reportCSVPath ) :
    backupCSVPath = self . __backupCSVPath = _backupFile ( reportCSVPath )
# Open report file
if self . __replaceReport :
    mode = "w"
else :
    mode = "a"
csv = self . __csvFileObj = open ( reportCSVPath , mode )
# If we are appending , add some blank line separators
if not self . __replaceReport and backupCSVPath :
    print >> csv
    print >> csv
# Print the column names
print >> csv , "jobID, " ,
print >> csv , "modelID, " ,
print >> csv , "status, " ,
print >> csv , "completionReason, " ,
print >> csv , "startTime, " ,
print >> csv , "endTime, " ,
print >> csv , "runtime(s), " ,
print >> csv , "expDesc, " ,
print >> csv , "numRecords, " ,
for key in self . __sortedVariableNames :
    print >> csv , "%s, " % key ,
for key in self . __sortedMetricsKeys :
    print >> csv , "%s, " % key ,
print >> csv


signature:
def getJobStatus ( self , workers ) :
    """Parameters :
 workers : If this job was launched outside of the nupic job engine , then this
 is an array of subprocess Popen instances , one for each worker
 retval : _ NupicJob . JobStatus instance"""

implementation:
jobInfo = self . JobStatus ( self . __nupicJobID , workers )
return jobInfo


signature:
def queryModelIDs ( self ) :
    """Queuries DB for model IDs of all currently instantiated models
 associated with this HyperSearch job .
 See also : _ iterModels ( )
 Parameters :
 retval : A sequence of Nupic modelIDs"""

implementation:
jobID = self . getJobID ( )
modelCounterPairs = _clientJobsDB ( ) . modelsGetUpdateCounters ( jobID )
modelIDs = tuple ( x [ 0 ] for x in modelCounterPairs )
return modelIDs


signature:
def makeSearchJobParamsDict ( cls , options , forRunning = False ) :
    """Constructs a dictionary of HyperSearch parameters suitable for converting
 to json and passing as the params argument to ClientJobsDAO . jobInsert ( )
 Parameters :
 options : NupicRunPermutations options dict
 forRunning : True if the params are for running a Hypersearch job ; False
 if params are for introspection only .
 retval : A dictionary of HyperSearch parameters for
 ClientJobsDAO . jobInsert ( )"""

implementation:
if options [ "searchMethod" ] == "v2" :
    hsVersion = "v2"
else :
    raise Exception ( "Unsupported search method: %r" % options [ "searchMethod" ] )
maxModels = options [ "maxPermutations" ]
if options [ "action" ] == "dryRun" and maxModels is None :
    maxModels = 1
useTerminators = options [ "useTerminators" ]
if useTerminators is None :
    params = { "hsVersion" : hsVersion , "maxModels" : maxModels , }
else :
    params = { "hsVersion" : hsVersion , "useTerminators" : useTerminators , "maxModels" : maxModels , }
if forRunning :
    params [ "persistentJobGUID" ] = str ( uuid . uuid1 ( ) )
if options [ "permutationsScriptPath" ] :
    params [ "permutationsPyFilename" ] = options [ "permutationsScriptPath" ]
elif options [ "expDescConfig" ] :
    params [ "description" ] = options [ "expDescConfig" ]
else :
    with open ( options [ "expDescJsonPath" ] , mode = "r" ) as fp :
        params [ "description" ] = json . load ( fp )
return params


signature:
def getOptimizationMetricInfo ( cls , searchJobParams ) :
    """Retrives the optimization key name and optimization function .
 Parameters :
 searchJobParams :
 Parameter for passing as the searchParams arg to
 Hypersearch constructor .
 retval : ( optimizationMetricKey , maximize )
 optimizationMetricKey : which report key to optimize for
 maximize : True if we should try and maximize the optimizeKey
 metric . False if we should minimize it ."""

implementation:
if searchJobParams [ "hsVersion" ] == "v2" :
    search = HypersearchV2 ( searchParams = searchJobParams )
else :
    raise RuntimeError ( "Unsupported hypersearch version \"%s\"" % ( searchJobParams [ "hsVersion" ] ) )
info = search . getOptimizationMetricInfo ( )
return info


signature:
def getModelDescription ( self ) :
    """Parameters :
 retval : Printable description of the model ."""

implementation:
params = self . __unwrapParams ( )
if "experimentName" in params :
    return params [ "experimentName" ]
else :
    paramSettings = self . getParamLabels ( )
    # Form a csv friendly string representation of this model
    items = [ ]
    for key , value in paramSettings . items ( ) :
        items . append ( "%s_%s" % ( key , value ) )
    return "." . join ( items )


signature:
def getParamLabels ( self ) :
    """Parameters :
 retval : a dictionary of model parameter labels . For each entry
 the key is the name of the parameter and the value
 is the value chosen for it ."""

implementation:
params = self . __unwrapParams ( )
# Hypersearch v2 stores the flattened parameter settings in " particleState "
if "particleState" in params :
    retval = dict ( )
    queue = [ ( pair , retval ) for pair in params [ "particleState" ] [ "varStates" ] . iteritems ( ) ]
    while len ( queue ) > 0 :
        pair , output = queue . pop ( )
        k , v = pair
        if ( "position" in v and "bestPosition" in v and "velocity" in v ) :
            output [ k ] = v [ "position" ]
        else :
            if k not in output :
                output [ k ] = dict ( )
            queue . extend ( ( pair , output [ k ] ) for pair in v . iteritems ( ) )
    return retval


signature:
def __unwrapParams ( self ) :
    """Unwraps self . _ _ rawInfo . params into the equivalent python dictionary
 and caches it in self . _ _ cachedParams . Returns the unwrapped params
 Parameters :
 retval : Model params dictionary as correpsonding to the json
 as returned in ClientJobsDAO . modelsInfo ( ) [ x ] . params"""

implementation:
if self . __cachedParams is None :
    self . __cachedParams = json . loads ( self . __rawInfo . params )
    assert self . __cachedParams is not None , "%s resulted in None" % self . __rawInfo . params
return self . __cachedParams


signature:
def getAllMetrics ( self ) :
    """Retrives a dictionary of metrics that combines all report and
 optimization metrics
 Parameters :
 retval : a dictionary of optimization metrics that were collected
 for the model ; an empty dictionary if there aren ' t any ."""

implementation:
result = self . getReportMetrics ( )
result . update ( self . getOptimizationMetrics ( ) )
return result


signature:
def __unwrapResults ( self ) :
    """Unwraps self . _ _ rawInfo . results and caches it in self . _ _ cachedResults ;
 Returns the unwrapped params
 Parameters :
 retval : ModelResults namedtuple instance"""

implementation:
if self . __cachedResults is None :
    if self . __rawInfo . results is not None :
        resultList = json . loads ( self . __rawInfo . results )
        assert len ( resultList ) == 2 , "Expected 2 elements, but got %s (%s)." % ( len ( resultList ) , resultList )
        self . __cachedResults = self . ModelResults ( reportMetrics = resultList [ 0 ] , optimizationMetrics = resultList [ 1 ] )
    else :
        self . __cachedResults = self . ModelResults ( reportMetrics = { } , optimizationMetrics = { } )
return self . __cachedResults


signature:
def getData ( self , n ) :
    """Returns the next n values for the distribution as a list ."""

implementation:
records = [ self . getNext ( ) for x in range ( n ) ]
return records


signature:
def getTerminationCallbacks ( self , terminationFunc ) :
    """Returns the periodic checks to see if the model should
 continue running .
 Parameters :
 terminationFunc : The function that will be called in the model main loop
 as a wrapper around this function . Must have a parameter
 called ' index '
 Returns : A list of PeriodicActivityRequest objects ."""

implementation:
activities = [ None ] * len ( ModelTerminator . _MILESTONES )
for index , ( iteration , _ ) in enumerate ( ModelTerminator . _MILESTONES ) :
    cb = functools . partial ( terminationFunc , index = index )
    activities [ index ] = PeriodicActivityRequest ( repeating = False , period = iteration , cb = cb )


signature:
def groupby2 ( * args ) :
    """Like itertools . groupby , with the following additions :
 - Supports multiple sequences . Instead of returning ( k , g ) , each iteration
 returns ( k , g0 , g1 , . . . ) , with one ` g ` for each input sequence . The value of
 each ` g ` is either a non - empty iterator or ` None ` .
 - It treats the value ` None ` as an empty sequence . So you can make subsequent
 calls to groupby2 on any ` g ` value .
 . . note : : Read up on groupby here :
 https : / / docs . python . org / dev / library / itertools . html # itertools . groupby
 : param args : ( list ) Parameters alternating between sorted lists and their
 respective key functions . The lists should be sorted with
 respect to their key function .
 : returns : ( tuple ) A n + 1 dimensional tuple , where the first element is the
 key of the iteration , and the other n entries are groups of
 objects that share this key . Each group corresponds to the an
 input sequence . ` groupby2 ` is a generator that returns a tuple
 for every iteration . If an input sequence has no members with
 the current key , None is returned in place of a generator ."""

implementation:
generatorList = [ ]
# list of each list ' s ( k , group ) tuples
if len ( args ) % 2 == 1 :
    raise ValueError ( "Must have a key function for every list." )
advanceList = [ ]
# populate above lists
for i in xrange ( 0 , len ( args ) , 2 ) :
    listn = args [ i ]
    fn = args [ i + 1 ]
    if listn is not None :
        generatorList . append ( groupby ( listn , fn ) )
        advanceList . append ( True )
        # start by advancing everyone .
    else :
        generatorList . append ( None )
        advanceList . append ( False )
n = len ( generatorList )
nextList = [ None ] * n
# while all lists aren ' t exhausted walk through each group in order
while True :
    for i in xrange ( n ) :
        if advanceList [ i ] :
            try :
                nextList [ i ] = generatorList [ i ] . next ( )
            except StopIteration :
                nextList [ i ] = None
    # no more values to process in any of the generators
    if all ( entry is None for entry in nextList ) :
        break
    # the minimum key value in the nextList
    minKeyVal = min ( nextVal [ 0 ] for nextVal in nextList if nextVal is not None )
    # populate the tuple to return based on minKeyVal
    retGroups = [ minKeyVal ]
    for i in xrange ( n ) :
        if nextList [ i ] is not None and nextList [ i ] [ 0 ] == minKeyVal :
            retGroups . append ( nextList [ i ] [ 1 ] )
            advanceList [ i ] = True
        else :
            advanceList [ i ] = False
            retGroups . append ( None )
    yield tuple ( retGroups )


signature:
def _openStream ( dataUrl , isBlocking , # pylint : disable = W0613
maxTimeout , # pylint : disable = W0613
bookmark , firstRecordIdx ) :
    """Open the underlying file stream
 This only supports ' file : / / ' prefixed paths .
 : returns : record stream instance
 : rtype : FileRecordStream"""

implementation:
filePath = dataUrl [ len ( FILE_PREF ) : ]
if not os . path . isabs ( filePath ) :
    filePath = os . path . join ( os . getcwd ( ) , filePath )
return FileRecordStream ( streamID = filePath , write = False , bookmark = bookmark , firstRecord = firstRecordIdx )


signature:
def getNextRecord ( self ) :
    """Returns combined data from all sources ( values only ) .
 : returns : None on EOF ; empty sequence on timeout ."""

implementation:
# Keep reading from the raw input till we get enough for an aggregated
# record
while True : # Reached EOF due to lastRow constraint ?
    if self . _sourceLastRecordIdx is not None and self . _recordStore . getNextRecordIdx ( ) >= self . _sourceLastRecordIdx :
        preAggValues = None
        # indicates EOF
        bookmark = self . _recordStore . getBookmark ( )
    else : # Get the raw record and bookmark
        preAggValues = self . _recordStore . getNextRecord ( )
        bookmark = self . _recordStore . getBookmark ( )
    if preAggValues == ( ) : # means timeout error occurred
        if self . _eofOnTimeout :
            preAggValues = None
            # act as if we got EOF
        else :
            return preAggValues
    # Timeout indicator
    self . _logger . debug ( 'Read source record #%d: %r' , self . _recordStore . getNextRecordIdx ( ) - 1 , preAggValues )
    # Perform aggregation
    ( fieldValues , aggBookmark ) = self . _aggregator . next ( preAggValues , bookmark )
    # Update the aggregated record bookmark if we got a real record back
    if fieldValues is not None :
        self . _aggBookmark = aggBookmark
    # Reached EOF ?
    if preAggValues is None and fieldValues is None :
        return None
    # Return it if we have a record
    if fieldValues is not None :
        break
# Do we need to re - order the fields in the record ?
if self . _needFieldsFiltering :
    values = [ ]
    srcDict = dict ( zip ( self . _recordStoreFieldNames , fieldValues ) )
    for name in self . _streamFieldNames :
        values . append ( srcDict [ name ] )
    fieldValues = values
# Write to debug output ?
if self . _writer is not None :
    self . _writer . appendRecord ( fieldValues )
self . _recordCount += 1
self . _logger . debug ( 'Returning aggregated record #%d from getNextRecord(): ' '%r. Bookmark: %r' , self . _recordCount - 1 , fieldValues , self . _aggBookmark )
return fieldValues


signature:
def getDataRowCount ( self ) :
    """Iterates through stream to calculate total records after aggregation .
 This will alter the bookmark state ."""

implementation:
inputRowCountAfterAggregation = 0
while True :
    record = self . getNextRecord ( )
    if record is None :
        return inputRowCountAfterAggregation
    inputRowCountAfterAggregation += 1
    if inputRowCountAfterAggregation > 10000 :
        raise RuntimeError ( 'No end of datastream found.' )


signature:
def getStats ( self ) :
    """TODO : This method needs to be enhanced to get the stats on the * aggregated *
 records .
 : returns : stats ( like min and max values of the fields ) ."""

implementation:
# The record store returns a dict of stats , each value in this dict is
# a list with one item per field of the record store
# ' min ' : [ f1 _ min , f2 _ min , f3 _ min ] ,
# ' max ' : [ f1 _ max , f2 _ max , f3 _ max ]
recordStoreStats = self . _recordStore . getStats ( )
# We need to convert each item to represent the fields of the * stream *
streamStats = dict ( )
for ( key , values ) in recordStoreStats . items ( ) :
    fieldStats = dict ( zip ( self . _recordStoreFieldNames , values ) )
    streamValues = [ ]
    for name in self . _streamFieldNames :
        streamValues . append ( fieldStats [ name ] )
    streamStats [ key ] = streamValues
return streamStats


signature:
def get ( self , number ) :
    """Return a pattern for a number .
 @ param number ( int ) Number of pattern
 @ return ( set ) Indices of on bits"""

implementation:
if not number in self . _patterns :
    raise IndexError ( "Invalid number" )
return self . _patterns [ number ]


signature:
def addNoise ( self , bits , amount ) :
    """Add noise to pattern .
 @ param bits ( set ) Indices of on bits
 @ param amount ( float ) Probability of switching an on bit with a random bit
 @ return ( set ) Indices of on bits in noisy pattern"""

implementation:
newBits = set ( )
for bit in bits :
    if self . _random . getReal64 ( ) < amount :
        newBits . add ( self . _random . getUInt32 ( self . _n ) )
    else :
        newBits . add ( bit )
return newBits


signature:
def numbersForBit ( self , bit ) :
    """Return the set of pattern numbers that match a bit .
 @ param bit ( int ) Index of bit
 @ return ( set ) Indices of numbers"""

implementation:
if bit >= self . _n :
    raise IndexError ( "Invalid bit" )
numbers = set ( )
for index , pattern in self . _patterns . iteritems ( ) :
    if bit in pattern :
        numbers . add ( index )
return numbers


signature:
def numberMapForBits ( self , bits ) :
    """Return a map from number to matching on bits ,
 for all numbers that match a set of bits .
 @ param bits ( set ) Indices of bits
 @ return ( dict ) Mapping from number = > on bits ."""

implementation:
numberMap = dict ( )
for bit in bits :
    numbers = self . numbersForBit ( bit )
    for number in numbers :
        if not number in numberMap :
            numberMap [ number ] = set ( )
        numberMap [ number ] . add ( bit )
return numberMap


signature:
def prettyPrintPattern ( self , bits , verbosity = 1 ) :
    """Pretty print a pattern .
 @ param bits ( set ) Indices of on bits
 @ param verbosity ( int ) Verbosity level
 @ return ( string ) Pretty - printed text"""

implementation:
numberMap = self . numberMapForBits ( bits )
text = ""
numberList = [ ]
numberItems = sorted ( numberMap . iteritems ( ) , key = lambda ( number , bits ) : len ( bits ) , reverse = True )
for number , bits in numberItems :
    if verbosity > 2 :
        strBits = [ str ( n ) for n in bits ]
        numberText = "{0} (bits: {1})" . format ( number , "," . join ( strBits ) )
    elif verbosity > 1 :
        numberText = "{0} ({1} bits)" . format ( number , len ( bits ) )
    else :
        numberText = str ( number )
    numberList . append ( numberText )
text += "[{0}]" . format ( ", " . join ( numberList ) )
return text


signature:
def _generate ( self ) :
    """Generates set of random patterns ."""

implementation:
candidates = np . array ( range ( self . _n ) , np . uint32 )
for i in xrange ( self . _num ) :
    self . _random . shuffle ( candidates )
    pattern = candidates [ 0 : self . _getW ( ) ]
    self . _patterns [ i ] = set ( pattern )


signature:
def _getW ( self ) :
    """Gets a value of ` w ` for use in generating a pattern ."""

implementation:
w = self . _w
if type ( w ) is list :
    return w [ self . _random . getUInt32 ( len ( w ) ) ]
else :
    return w


signature:
def _generate ( self ) :
    """Generates set of consecutive patterns ."""

implementation:
n = self . _n
w = self . _w
assert type ( w ) is int , "List for w not supported"
for i in xrange ( n / w ) :
    pattern = set ( xrange ( i * w , ( i + 1 ) * w ) )
    self . _patterns [ i ] = pattern


signature:
def compute ( self , recordNum , patternNZ , classification , learn , infer ) :
    """Process one input sample .
 This method is called by outer loop code outside the nupic - engine . We
 use this instead of the nupic engine compute ( ) because our inputs and
 outputs aren ' t fixed size vectors of reals .
 : param recordNum : Record number of this input pattern . Record numbers
 normally increase sequentially by 1 each time unless there are missing
 records in the dataset . Knowing this information insures that we don ' t get
 confused by missing records .
 : param patternNZ : List of the active indices from the output below . When the
 input is from TemporalMemory , this list should be the indices of the
 active cells .
 : param classification : Dict of the classification information where :
 - bucketIdx : list of indices of the encoder bucket
 - actValue : list of actual values going into the encoder
 Classification could be None for inference mode .
 : param learn : ( bool ) if true , learn this sample
 : param infer : ( bool ) if true , perform inference
 : return : Dict containing inference results , there is one entry for each
 step in self . steps , where the key is the number of steps , and
 the value is an array containing the relative likelihood for
 each bucketIdx starting from bucketIdx 0.
 There is also an entry containing the average actual value to
 use for each bucket . The key is ' actualValues ' .
 for example :
 . . code - block : : python
 {1 : [ 0.1 , 0.3 , 0.2 , 0.7 ] ,
 4 : [ 0.2 , 0.4 , 0.3 , 0.5 ] ,
 ' actualValues ' : [ 1.5 , 3,5 , 5,5 , 7.6 ] ,"""

implementation:
if self . verbosity >= 1 :
    print "  learn:" , learn
    print "  recordNum:" , recordNum
    print "  patternNZ (%d):" % len ( patternNZ ) , patternNZ
    print "  classificationIn:" , classification
# ensures that recordNum increases monotonically
if len ( self . _patternNZHistory ) > 0 :
    if recordNum < self . _patternNZHistory [ - 1 ] [ 0 ] :
        raise ValueError ( "the record number has to increase monotonically" )
# Store pattern in our history if this is a new record
if len ( self . _patternNZHistory ) == 0 or recordNum > self . _patternNZHistory [ - 1 ] [ 0 ] :
    self . _patternNZHistory . append ( ( recordNum , patternNZ ) )
# To allow multi - class classification , we need to be able to run learning
# without inference being on . So initialize retval outside
# of the inference block .
retval = { }
# Update maxInputIdx and augment weight matrix with zero padding
if max ( patternNZ ) > self . _maxInputIdx :
    newMaxInputIdx = max ( patternNZ )
    for nSteps in self . steps :
        self . _weightMatrix [ nSteps ] = numpy . concatenate ( ( self . _weightMatrix [ nSteps ] , numpy . zeros ( shape = ( newMaxInputIdx - self . _maxInputIdx , self . _maxBucketIdx + 1 ) ) ) , axis = 0 )
    self . _maxInputIdx = int ( newMaxInputIdx )
# Get classification info
if classification is not None :
    if type ( classification [ "bucketIdx" ] ) is not list :
        bucketIdxList = [ classification [ "bucketIdx" ] ]
        actValueList = [ classification [ "actValue" ] ]
        numCategory = 1
    else :
        bucketIdxList = classification [ "bucketIdx" ]
        actValueList = classification [ "actValue" ]
        numCategory = len ( classification [ "bucketIdx" ] )
else :
    if learn :
        raise ValueError ( "classification cannot be None when learn=True" )
    actValueList = None
    bucketIdxList = None
# Inference :
# For each active bit in the activationPattern , get the classification
# votes
if infer :
    retval = self . infer ( patternNZ , actValueList )
if learn and classification [ "bucketIdx" ] is not None :
    for categoryI in range ( numCategory ) :
        bucketIdx = bucketIdxList [ categoryI ]
        actValue = actValueList [ categoryI ]
        # Update maxBucketIndex and augment weight matrix with zero padding
        if bucketIdx > self . _maxBucketIdx :
            for nSteps in self . steps :
                self . _weightMatrix [ nSteps ] = numpy . concatenate ( ( self . _weightMatrix [ nSteps ] , numpy . zeros ( shape = ( self . _maxInputIdx + 1 , bucketIdx - self . _maxBucketIdx ) ) ) , axis = 1 )
            self . _maxBucketIdx = int ( bucketIdx )
        # Update rolling average of actual values if it ' s a scalar . If it ' s
        # not , it must be a category , in which case each bucket only ever
        # sees one category so we don ' t need a running average .
        while self . _maxBucketIdx > len ( self . _actualValues ) - 1 :
            self . _actualValues . append ( None )
        if self . _actualValues [ bucketIdx ] is None :
            self . _actualValues [ bucketIdx ] = actValue
        else :
            if ( isinstance ( actValue , int ) or isinstance ( actValue , float ) or isinstance ( actValue , long ) ) :
                self . _actualValues [ bucketIdx ] = ( ( 1.0 - self . actValueAlpha ) * self . _actualValues [ bucketIdx ] + self . actValueAlpha * actValue )
            else :
                self . _actualValues [ bucketIdx ] = actValue
    for ( learnRecordNum , learnPatternNZ ) in self . _patternNZHistory :
        error = self . _calculateError ( recordNum , bucketIdxList )
        nSteps = recordNum - learnRecordNum
        if nSteps in self . steps :
            for bit in learnPatternNZ :
                self . _weightMatrix [ nSteps ] [ bit , : ] += self . alpha * error [ nSteps ]
# Verbose print
if infer and self . verbosity >= 1 :
    print "  inference: combined bucket likelihoods:"
    print "    actual bucket values:" , retval [ "actualValues" ]
    for ( nSteps , votes ) in retval . items ( ) :
        if nSteps == "actualValues" :
            continue
        print "    %d steps: " % ( nSteps ) , _pFormatArray ( votes )
        bestBucketIdx = votes . argmax ( )
        print ( "      most likely bucket idx: " "%d, value: %s" % ( bestBucketIdx , retval [ "actualValues" ] [ bestBucketIdx ] ) )
    print
return retval


signature:
def infer ( self , patternNZ , actValueList ) :
    """Return the inference value from one input sample . The actual
 learning happens in compute ( ) .
 : 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
 : return : 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 bucketIdx 0.
 for example :
 . . code - block : : python
 { ' 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:
# Return value dict . For buckets which we don ' t have an actual value
# for yet , just plug in any valid actual value . It doesn ' t matter what
# we use because that bucket won ' t have non - zero likelihood anyways .
# NOTE : If doing 0 - step prediction , we shouldn ' t use any knowledge
# of the classification input during inference .
if self . steps [ 0 ] == 0 or actValueList is None :
    defaultValue = 0
else :
    defaultValue = actValueList [ 0 ]
actValues = [ x if x is not None else defaultValue for x in self . _actualValues ]
retval = { "actualValues" : actValues }
for nSteps in self . steps :
    predictDist = self . inferSingleStep ( patternNZ , self . _weightMatrix [ nSteps ] )
    retval [ nSteps ] = predictDist
return retval


signature:
def inferSingleStep ( self , patternNZ , weightMatrix ) :
    """Perform inference for a single step . Given an SDR input and a weight
 matrix , return a predicted distribution .
 : param patternNZ : list of the active indices from the output below
 : param weightMatrix : numpy array of the weight matrix
 : return : numpy array of the predicted class label distribution"""

implementation:
outputActivation = weightMatrix [ patternNZ ] . sum ( axis = 0 )
# softmax normalization
outputActivation = outputActivation - numpy . max ( outputActivation )
expOutputActivation = numpy . exp ( outputActivation )
predictDist = expOutputActivation / numpy . sum ( expOutputActivation )
return predictDist


signature:
def _calculateError ( self , recordNum , bucketIdxList ) :
    """Calculate error signal
 : param bucketIdxList : list of encoder buckets
 : return : dict containing error . The key is the number of steps
 The value is a numpy array of error at the output layer"""

implementation:
error = dict ( )
targetDist = numpy . zeros ( self . _maxBucketIdx + 1 )
numCategories = len ( bucketIdxList )
for bucketIdx in bucketIdxList :
    targetDist [ bucketIdx ] = 1.0 / numCategories
for ( learnRecordNum , learnPatternNZ ) in self . _patternNZHistory :
    nSteps = recordNum - learnRecordNum
    if nSteps in self . steps :
        predictDist = self . inferSingleStep ( learnPatternNZ , self . _weightMatrix [ nSteps ] )
        error [ nSteps ] = targetDist - predictDist
return error