signature:
def generateRandomInput ( numRecords , elemSize = 400 , numSet = 42 ) :
    """Generates a set of input record
 Params :
 numRecords - how many records to generate
 elemSize - the size of each record ( num 0s or 1s )
 numSet - how many 1s in each record
 Returns : a list of inputs"""

implementation:
inputs = [ ]
for _ in xrange ( numRecords ) :
    input = np . zeros ( elemSize , dtype = realDType )
    for _ in range ( 0 , numSet ) :
        ind = np . random . random_integers ( 0 , elemSize - 1 , 1 ) [ 0 ]
        input [ ind ] = 1
    while abs ( input . sum ( ) - numSet ) > 0.1 :
        ind = np . random . random_integers ( 0 , elemSize - 1 , 1 ) [ 0 ]
        input [ ind ] = 1
    inputs . append ( input )
return inputs


signature:
def appendInputWithSimilarValues ( inputs ) :
    """Creates an ' one - off ' record for each record in the inputs . Appends new
 records to the same inputs list ."""

implementation:
numInputs = len ( inputs )
for i in xrange ( numInputs ) :
    input = inputs [ i ]
    for j in xrange ( len ( input ) - 1 ) :
        if input [ j ] == 1 and input [ j + 1 ] == 0 :
            newInput = copy . deepcopy ( input )
            newInput [ j ] = 0
            newInput [ j + 1 ] = 1
            inputs . append ( newInput )
            break


signature:
def appendInputWithNSimilarValues ( inputs , numNear = 10 ) :
    """Creates a neighboring record for each record in the inputs and adds
 new records at the end of the inputs list"""

implementation:
numInputs = len ( inputs )
skipOne = False
for i in xrange ( numInputs ) :
    input = inputs [ i ]
    numChanged = 0
    newInput = copy . deepcopy ( input )
    for j in xrange ( len ( input ) - 1 ) :
        if skipOne :
            skipOne = False
            continue
        if input [ j ] == 1 and input [ j + 1 ] == 0 :
            newInput [ j ] = 0
            newInput [ j + 1 ] = 1
            inputs . append ( newInput )
            newInput = copy . deepcopy ( newInput )
            # print input
            # print newInput
            numChanged += 1
            skipOne = True
            if numChanged == numNear :
                break


signature:
def modifyBits ( inputVal , maxChanges ) :
    """Modifies up to maxChanges number of bits in the inputVal"""

implementation:
changes = np . random . random_integers ( 0 , maxChanges , 1 ) [ 0 ]
if changes == 0 :
    return inputVal
inputWidth = len ( inputVal )
whatToChange = np . random . random_integers ( 0 , 41 , changes )
runningIndex = - 1
numModsDone = 0
for i in xrange ( inputWidth ) :
    if numModsDone >= changes :
        break
    if inputVal [ i ] == 1 :
        runningIndex += 1
        if runningIndex in whatToChange :
            if i != 0 and inputVal [ i - 1 ] == 0 :
                inputVal [ i - 1 ] = 1
                inputVal [ i ] = 0
return inputVal


signature:
def getRandomWithMods ( inputSpace , maxChanges ) :
    """Returns a random selection from the inputSpace with randomly modified
 up to maxChanges number of bits ."""

implementation:
size = len ( inputSpace )
ind = np . random . random_integers ( 0 , size - 1 , 1 ) [ 0 ]
value = copy . deepcopy ( inputSpace [ ind ] )
if maxChanges == 0 :
    return value
return modifyBits ( value , maxChanges )


signature:
def createEncoder ( ) :
    """Creates and returns a # MultiEncoder including a ScalarEncoder for
 energy consumption and a DateEncoder for the time of the day .
 @ see nupic / encoders / _ _ init _ _ . py for type to file - name mapping
 @ see nupic / encoders for encoder source files"""

implementation:
encoder = MultiEncoder ( )
encoder . addMultipleEncoders ( { "consumption" : { "fieldname" : u"consumption" , "type" : "ScalarEncoder" , "name" : u"consumption" , "minval" : 0.0 , "maxval" : 100.0 , "clipInput" : True , "w" : 21 , "n" : 500 } , "timestamp_timeOfDay" : { "fieldname" : u"timestamp" , "type" : "DateEncoder" , "name" : u"timestamp_timeOfDay" , "timeOfDay" : ( 21 , 9.5 ) } } )
return encoder


signature:
def createRecordSensor ( network , name , dataSource ) :
    """Creates a RecordSensor region that allows us to specify a file record
 stream as the input source ."""

implementation:
# Specific type of region . Possible options can be found in / nupic / regions /
regionType = "py.RecordSensor"
# Creates a json from specified dictionary .
regionParams = json . dumps ( { "verbosity" : _VERBOSITY } )
network . addRegion ( name , regionType , regionParams )
# getSelf returns the actual region , instead of a region wrapper
sensorRegion = network . regions [ name ] . getSelf ( )
# Specify how RecordSensor encodes input values
sensorRegion . encoder = createEncoder ( )
# Specify which sub - encoder should be used for " actValueOut "
network . regions [ name ] . setParameter ( "predictedField" , "consumption" )
# Specify the dataSource as a file record stream instance
sensorRegion . dataSource = dataSource
return sensorRegion


signature:
def createNetwork ( dataSource ) :
    """Creates and returns a new Network with a sensor region reading data from
 ' dataSource ' . There are two hierarchical levels , each with one SP and one TM .
 @ param dataSource - A RecordStream containing the input data
 @ returns a Network ready to run"""

implementation:
network = Network ( )
# Create and add a record sensor and a SP region
sensor = createRecordSensor ( network , name = _RECORD_SENSOR , dataSource = dataSource )
createSpatialPooler ( network , name = _L1_SPATIAL_POOLER , inputWidth = sensor . encoder . getWidth ( ) )
# Link the SP region to the sensor input
linkType = "UniformLink"
linkParams = ""
network . link ( _RECORD_SENSOR , _L1_SPATIAL_POOLER , linkType , linkParams )
# Create and add a TM region
l1temporalMemory = createTemporalMemory ( network , _L1_TEMPORAL_MEMORY )
# Link SP region to TM region in the feedforward direction
network . link ( _L1_SPATIAL_POOLER , _L1_TEMPORAL_MEMORY , linkType , linkParams )
# Add a classifier
classifierParams = { # Learning rate . Higher values make it adapt faster .
'alpha' : 0.005 , # A comma separated list of the number of steps the
# classifier predicts in the future . The classifier will
# learn predictions of each order specified .
'steps' : '1' , # The specific implementation of the classifier to use
# See SDRClassifierFactory # create for options
'implementation' : 'py' , # Diagnostic output verbosity control ;
# 0 : silent ; [ 1 . . 6 ] : increasing levels of verbosity
'verbosity' : 0 }
l1Classifier = network . addRegion ( _L1_CLASSIFIER , "py.SDRClassifierRegion" , json . dumps ( classifierParams ) )
l1Classifier . setParameter ( 'inferenceMode' , True )
l1Classifier . setParameter ( 'learningMode' , True )
network . link ( _L1_TEMPORAL_MEMORY , _L1_CLASSIFIER , linkType , linkParams , srcOutput = "bottomUpOut" , destInput = "bottomUpIn" )
network . link ( _RECORD_SENSOR , _L1_CLASSIFIER , linkType , linkParams , srcOutput = "categoryOut" , destInput = "categoryIn" )
network . link ( _RECORD_SENSOR , _L1_CLASSIFIER , linkType , linkParams , srcOutput = "bucketIdxOut" , destInput = "bucketIdxIn" )
network . link ( _RECORD_SENSOR , _L1_CLASSIFIER , linkType , linkParams , srcOutput = "actValueOut" , destInput = "actValueIn" )
# Second Level
l2inputWidth = l1temporalMemory . getSelf ( ) . getOutputElementCount ( "bottomUpOut" )
createSpatialPooler ( network , name = _L2_SPATIAL_POOLER , inputWidth = l2inputWidth )
network . link ( _L1_TEMPORAL_MEMORY , _L2_SPATIAL_POOLER , linkType , linkParams )
createTemporalMemory ( network , _L2_TEMPORAL_MEMORY )
network . link ( _L2_SPATIAL_POOLER , _L2_TEMPORAL_MEMORY , linkType , linkParams )
l2Classifier = network . addRegion ( _L2_CLASSIFIER , "py.SDRClassifierRegion" , json . dumps ( classifierParams ) )
l2Classifier . setParameter ( 'inferenceMode' , True )
l2Classifier . setParameter ( 'learningMode' , True )
network . link ( _L2_TEMPORAL_MEMORY , _L2_CLASSIFIER , linkType , linkParams , srcOutput = "bottomUpOut" , destInput = "bottomUpIn" )
network . link ( _RECORD_SENSOR , _L2_CLASSIFIER , linkType , linkParams , srcOutput = "categoryOut" , destInput = "categoryIn" )
network . link ( _RECORD_SENSOR , _L2_CLASSIFIER , linkType , linkParams , srcOutput = "bucketIdxOut" , destInput = "bucketIdxIn" )
network . link ( _RECORD_SENSOR , _L2_CLASSIFIER , linkType , linkParams , srcOutput = "actValueOut" , destInput = "actValueIn" )
return network


signature:
def runNetwork ( network , numRecords , writer ) :
    """Runs specified Network writing the ensuing anomaly
 scores to writer .
 @ param network : The Network instance to be run
 @ param writer : A csv . writer used to write to output file ."""

implementation:
sensorRegion = network . regions [ _RECORD_SENSOR ]
l1SpRegion = network . regions [ _L1_SPATIAL_POOLER ]
l1TpRegion = network . regions [ _L1_TEMPORAL_MEMORY ]
l1Classifier = network . regions [ _L1_CLASSIFIER ]
l2SpRegion = network . regions [ _L2_SPATIAL_POOLER ]
l2TpRegion = network . regions [ _L2_TEMPORAL_MEMORY ]
l2Classifier = network . regions [ _L2_CLASSIFIER ]
l1PreviousPredictedColumns = [ ]
l2PreviousPredictedColumns = [ ]
l1PreviousPrediction = None
l2PreviousPrediction = None
l1ErrorSum = 0.0
l2ErrorSum = 0.0
for record in xrange ( numRecords ) : # Run the network for a single iteration
    network . run ( 1 )
    actual = float ( sensorRegion . getOutputData ( "actValueOut" ) [ 0 ] )
    l1Predictions = l1Classifier . getOutputData ( "actualValues" )
    l1Probabilities = l1Classifier . getOutputData ( "probabilities" )
    l1Prediction = l1Predictions [ l1Probabilities . argmax ( ) ]
    if l1PreviousPrediction is not None :
        l1ErrorSum += math . fabs ( l1PreviousPrediction - actual )
    l1PreviousPrediction = l1Prediction
    l2Predictions = l2Classifier . getOutputData ( "actualValues" )
    l2Probabilities = l2Classifier . getOutputData ( "probabilities" )
    l2Prediction = l2Predictions [ l2Probabilities . argmax ( ) ]
    if l2PreviousPrediction is not None :
        l2ErrorSum += math . fabs ( l2PreviousPrediction - actual )
    l2PreviousPrediction = l2Prediction
    l1AnomalyScore = l1TpRegion . getOutputData ( "anomalyScore" ) [ 0 ]
    l2AnomalyScore = l2TpRegion . getOutputData ( "anomalyScore" ) [ 0 ]
    # Write record number , actualInput , and anomaly scores
    writer . writerow ( ( record , actual , l1PreviousPrediction , l1AnomalyScore , l2PreviousPrediction , l2AnomalyScore ) )
    # Store the predicted columns for the next timestep
    l1PredictedColumns = l1TpRegion . getOutputData ( "topDownOut" ) . nonzero ( ) [ 0 ]
    l1PreviousPredictedColumns = copy . deepcopy ( l1PredictedColumns )
    l2PredictedColumns = l2TpRegion . getOutputData ( "topDownOut" ) . nonzero ( ) [ 0 ]
    l2PreviousPredictedColumns = copy . deepcopy ( l2PredictedColumns )
# Output absolute average error for each level
if numRecords > 1 :
    print "L1 ave abs class. error: %f" % ( l1ErrorSum / ( numRecords - 1 ) )
    print "L2 ave abs class. error: %f" % ( l2ErrorSum / ( numRecords - 1 ) )


signature:
def clean ( s ) :
    """Removes trailing whitespace on each line ."""

implementation:
lines = [ l . rstrip ( ) for l in s . split ( '\n' ) ]
return '\n' . join ( lines )


signature:
def update ( self , results ) :
    """Compute the new metrics values , given the next inference / ground - truth values
 : param results : ( : class : ` ~ nupic . frameworks . opf . opf _ utils . ModelResult ` )
 object that was computed during the last iteration of the model .
 : returns : ( dict ) where each key is the metric - name , and the values are
 it scalar value ."""

implementation:
# print " \ n \ n - - - - - "
# print " Model results : \ nrawInput : % s \ ninferences : % s " % \
# ( pprint . pformat ( results . rawInput ) , pprint . pformat ( results . inferences ) )
self . _addResults ( results )
if not self . __metricSpecs or self . __currentInference is None :
    return { }
metricResults = { }
for metric , spec , label in zip ( self . __metrics , self . __metricSpecs , self . __metricLabels ) :
    inferenceElement = spec . inferenceElement
    field = spec . field
    groundTruth = self . _getGroundTruth ( inferenceElement )
    inference = self . _getInference ( inferenceElement )
    rawRecord = self . _getRawGroundTruth ( )
    result = self . __currentResult
    if field :
        if type ( inference ) in ( list , tuple ) :
            if field in self . __fieldNameIndexMap : # NOTE : If the predicted field is not fed in at the bottom , we
            # won ' t have it in our fieldNameIndexMap
                fieldIndex = self . __fieldNameIndexMap [ field ]
                inference = inference [ fieldIndex ]
            else :
                inference = None
        if groundTruth is not None :
            if type ( groundTruth ) in ( list , tuple ) :
                if field in self . __fieldNameIndexMap : # NOTE : If the predicted field is not fed in at the bottom , we
                # won ' t have it in our fieldNameIndexMap
                    fieldIndex = self . __fieldNameIndexMap [ field ]
                    groundTruth = groundTruth [ fieldIndex ]
                else :
                    groundTruth = None
            else : # groundTruth could be a dict based off of field names
                groundTruth = groundTruth [ field ]
    metric . addInstance ( groundTruth = groundTruth , prediction = inference , record = rawRecord , result = result )
    metricResults [ label ] = metric . getMetric ( ) [ 'value' ]
return metricResults


signature:
def getMetrics ( self ) :
    """Gets the current metric values
 : returns : ( dict ) where each key is the metric - name , and the values are
 it scalar value . Same as the output of
 : meth : ` ~ nupic . frameworks . opf . prediction _ metrics _ manager . MetricsManager . update `"""

implementation:
result = { }
for metricObj , label in zip ( self . __metrics , self . __metricLabels ) :
    value = metricObj . getMetric ( )
    result [ label ] = value [ 'value' ]
return result


signature:
def getMetricDetails ( self , metricLabel ) :
    """Gets detailed info about a given metric , in addition to its value . This
 may including any statistics or auxilary data that are computed for a given
 metric .
 : param metricLabel : ( string ) label of the given metric ( see
 : class : ` ~ nupic . frameworks . opf . metrics . MetricSpec ` )
 : returns : ( dict ) of metric information , as returned by
 : meth : ` nupic . frameworks . opf . metrics . MetricsIface . getMetric ` ."""

implementation:
try :
    metricIndex = self . __metricLabels . index ( metricLabel )
except IndexError :
    return None
return self . __metrics [ metricIndex ] . getMetric ( )


signature:
def _addResults ( self , results ) :
    """Stores the current model results in the manager ' s internal store
 Parameters :
 results : A ModelResults object that contains the current timestep ' s
 input / inferences"""

implementation:
# If the model potentially has temporal inferences .
if self . __isTemporal :
    shiftedInferences = self . __inferenceShifter . shift ( results ) . inferences
    self . __currentResult = copy . deepcopy ( results )
    self . __currentResult . inferences = shiftedInferences
    self . __currentInference = shiftedInferences
# The current model has no temporal inferences .
else :
    self . __currentResult = copy . deepcopy ( results )
    self . __currentInference = copy . deepcopy ( results . inferences )
# Save the current ground - truth results
self . __currentGroundTruth = copy . deepcopy ( results )


signature:
def _getGroundTruth ( self , inferenceElement ) :
    """Get the actual value for this field
 Parameters :
 sensorInputElement : The inference element ( part of the inference ) that
 is being used for this metric"""

implementation:
sensorInputElement = InferenceElement . getInputElement ( inferenceElement )
if sensorInputElement is None :
    return None
return getattr ( self . __currentGroundTruth . sensorInput , sensorInputElement )


signature:
def __constructMetricsModules ( self , metricSpecs ) :
    """Creates the required metrics modules
 Parameters :
 metricSpecs :
 A sequence of MetricSpec objects that specify which metric modules to
 instantiate"""

implementation:
if not metricSpecs :
    return
self . __metricSpecs = metricSpecs
for spec in metricSpecs :
    if not InferenceElement . validate ( spec . inferenceElement ) :
        raise ValueError ( "Invalid inference element for metric spec: %r" % spec )
    self . __metrics . append ( metrics . getModule ( spec ) )
    self . __metricLabels . append ( spec . getLabel ( ) )


signature:
def _generateSimple ( filename = "simple.csv" , numSequences = 1 , elementsPerSeq = 3 , numRepeats = 10 ) :
    """Generate a simple dataset . This contains a bunch of non - overlapping
 sequences .
 At the end of the dataset , we introduce missing records so that test
 code can insure that the model didn ' t get confused by them .
 Parameters :
 filename : name of the file to produce , including extension . It will
 be created in a ' datasets ' sub - directory within the
 directory containing this script .
 numSequences : how many sequences to generate
 elementsPerSeq : length of each sequence
 numRepeats : how many times to repeat each sequence in the output"""

implementation:
# Create the output file
scriptDir = os . path . dirname ( __file__ )
pathname = os . path . join ( scriptDir , 'datasets' , filename )
print "Creating %s..." % ( pathname )
fields = [ ( 'timestamp' , 'datetime' , 'T' ) , ( 'field1' , 'string' , '' ) , ( 'field2' , 'float' , '' ) ]
outFile = FileRecordStream ( pathname , write = True , fields = fields )
# Create the sequences
sequences = [ ]
for i in range ( numSequences ) :
    seq = [ x for x in range ( i * elementsPerSeq , ( i + 1 ) * elementsPerSeq ) ]
    sequences . append ( seq )
# Write out the sequences in random order
seqIdxs = [ ]
for i in range ( numRepeats ) :
    seqIdxs += range ( numSequences )
random . shuffle ( seqIdxs )
# Put 1 hour between each record
timestamp = datetime . datetime ( year = 2012 , month = 1 , day = 1 , hour = 0 , minute = 0 , second = 0 )
timeDelta = datetime . timedelta ( hours = 1 )
# Write out the sequences without missing records
for seqIdx in seqIdxs :
    seq = sequences [ seqIdx ]
    for x in seq :
        outFile . appendRecord ( [ timestamp , str ( x ) , x ] )
        timestamp += timeDelta
# Now , write some out with missing records
for seqIdx in seqIdxs :
    seq = sequences [ seqIdx ]
    for i , x in enumerate ( seq ) :
        if i != 1 :
            outFile . appendRecord ( [ timestamp , str ( x ) , x ] )
        timestamp += timeDelta
for seqIdx in seqIdxs :
    seq = sequences [ seqIdx ]
    for i , x in enumerate ( seq ) :
        if i != 1 :
            outFile . appendRecord ( [ timestamp , str ( x ) , x ] )
        timestamp += timeDelta
# Write out some more of the sequences * without * missing records
for seqIdx in seqIdxs :
    seq = sequences [ seqIdx ]
    for x in seq :
        outFile . appendRecord ( [ timestamp , str ( x ) , x ] )
        timestamp += timeDelta
outFile . close ( )


signature:
def shift ( self , modelResult ) :
    """Shift the model result and return the new instance .
 Queues up the T ( i + 1 ) prediction value and emits a T ( i )
 input / prediction pair , if possible . E . g . , if the previous T ( i - 1)
 iteration was learn - only , then we would not have a T ( i ) prediction in our
 FIFO and would not be able to emit a meaningful input / prediction pair .
 : param modelResult : A : class : ` ~ . nupic . frameworks . opf . opf _ utils . ModelResult `
 instance to shift .
 : return : A : class : ` ~ . nupic . frameworks . opf . opf _ utils . ModelResult ` instance that
 has been shifted"""

implementation:
inferencesToWrite = { }
if self . _inferenceBuffer is None :
    maxDelay = InferenceElement . getMaxDelay ( modelResult . inferences )
    self . _inferenceBuffer = collections . deque ( maxlen = maxDelay + 1 )
self . _inferenceBuffer . appendleft ( copy . deepcopy ( modelResult . inferences ) )
for inferenceElement , inference in modelResult . inferences . iteritems ( ) :
    if isinstance ( inference , dict ) :
        inferencesToWrite [ inferenceElement ] = { }
        for key , _ in inference . iteritems ( ) :
            delay = InferenceElement . getTemporalDelay ( inferenceElement , key )
            if len ( self . _inferenceBuffer ) > delay :
                prevInference = self . _inferenceBuffer [ delay ] [ inferenceElement ] [ key ]
                inferencesToWrite [ inferenceElement ] [ key ] = prevInference
            else :
                inferencesToWrite [ inferenceElement ] [ key ] = None
    else :
        delay = InferenceElement . getTemporalDelay ( inferenceElement )
        if len ( self . _inferenceBuffer ) > delay :
            inferencesToWrite [ inferenceElement ] = ( self . _inferenceBuffer [ delay ] [ inferenceElement ] )
        else :
            if type ( inference ) in ( list , tuple ) :
                inferencesToWrite [ inferenceElement ] = [ None ] * len ( inference )
            else :
                inferencesToWrite [ inferenceElement ] = None
shiftedResult = ModelResult ( rawInput = modelResult . rawInput , sensorInput = modelResult . sensorInput , inferences = inferencesToWrite , metrics = modelResult . metrics , predictedFieldIdx = modelResult . predictedFieldIdx , predictedFieldName = modelResult . predictedFieldName )
return shiftedResult


signature:
def generateStats ( filename , maxSamples = None , ) :
    """Collect statistics for each of the fields in the user input data file and
 return a stats dict object .
 Parameters :
 filename : The path and name of the data file .
 maxSamples : Upper bound on the number of rows to be processed
 retval : A dictionary of dictionaries . The top level keys are the
 field names and the corresponding values are the statistics
 collected for the individual file .
 Example :
 ' consumption ' : { ' min ' : 0 , ' max ' : 90 , ' mean ' : 50 , . . . } ,
 ' gym ' : { ' numDistinctCategories ' : 10 , . . . } ,"""

implementation:
# Mapping from field type to stats collector object
statsCollectorMapping = { 'float' : FloatStatsCollector , 'int' : IntStatsCollector , 'string' : StringStatsCollector , 'datetime' : DateTimeStatsCollector , 'bool' : BoolStatsCollector , }
filename = resource_filename ( "nupic.datafiles" , filename )
print "*" * 40
print "Collecting statistics for file:'%s'" % ( filename , )
dataFile = FileRecordStream ( filename )
# Initialize collector objects
# statsCollectors list holds statsCollector objects for each field
statsCollectors = [ ]
for fieldName , fieldType , fieldSpecial in dataFile . getFields ( ) : # Find the corresponding stats collector for each field based on field type
# and intialize an instance
    statsCollector = statsCollectorMapping [ fieldType ] ( fieldName , fieldType , fieldSpecial )
    statsCollectors . append ( statsCollector )
# Now collect the stats
if maxSamples is None :
    maxSamples = 500000
for i in xrange ( maxSamples ) :
    record = dataFile . getNextRecord ( )
    if record is None :
        break
    for i , value in enumerate ( record ) :
        statsCollectors [ i ] . addValue ( value )
# stats dict holds the statistics for each field
stats = { }
for statsCollector in statsCollectors :
    statsCollector . getStats ( stats )
# We don ' t want to include reset field in permutations
# TODO : handle reset field in a clean way
if dataFile . getResetFieldIdx ( ) is not None :
    resetFieldName , _ , _ = dataFile . getFields ( ) [ dataFile . reset ]
    stats . pop ( resetFieldName )
if VERBOSITY > 0 :
    pprint . pprint ( stats )
return stats


signature:
def getStats ( self , stats ) :
    """Override of getStats ( ) in BaseStatsCollector
 stats : A dictionary where all the stats are
 outputted"""

implementation:
BaseStatsCollector . getStats ( self , stats )
sortedNumberList = sorted ( self . valueList )
listLength = len ( sortedNumberList )
min = sortedNumberList [ 0 ]
max = sortedNumberList [ - 1 ]
mean = numpy . mean ( self . valueList )
median = sortedNumberList [ int ( 0.5 * listLength ) ]
percentile1st = sortedNumberList [ int ( 0.01 * listLength ) ]
percentile99th = sortedNumberList [ int ( 0.99 * listLength ) ]
differenceList = [ ( cur - prev ) for prev , cur in itertools . izip ( list ( self . valueSet ) [ : - 1 ] , list ( self . valueSet ) [ 1 : ] ) ]
if min > max :
    print self . fieldname , min , max , '-----'
meanResolution = numpy . mean ( differenceList )
stats [ self . fieldname ] [ 'min' ] = min
stats [ self . fieldname ] [ 'max' ] = max
stats [ self . fieldname ] [ 'mean' ] = mean
stats [ self . fieldname ] [ 'median' ] = median
stats [ self . fieldname ] [ 'percentile1st' ] = percentile1st
stats [ self . fieldname ] [ 'percentile99th' ] = percentile99th
stats [ self . fieldname ] [ 'meanResolution' ] = meanResolution
# TODO : Right now , always pass the data along .
# This is used for data - dependent encoders .
passData = True
if passData :
    stats [ self . fieldname ] [ 'data' ] = self . valueList
if VERBOSITY > 2 :
    print '--'
    print "Statistics:"
    print "min:" , min
    print "max:" , max
    print "mean:" , mean
    print "median:" , median
    print "1st percentile :" , percentile1st
    print "99th percentile:" , percentile99th
    print '--'
    print "Resolution:"
    print "Mean Resolution:" , meanResolution
if VERBOSITY > 3 :
    print '--'
    print "Histogram:"
    counts , bins = numpy . histogram ( self . valueList , new = True )
    print "Counts:" , counts . tolist ( )
    print "Bins:" , bins . tolist ( )


signature:
def main ( ) :
    """Run according to options in sys . argv and diff classifiers ."""

implementation:
initLogging ( verbose = True )
# Initialize PRNGs
initExperimentPrng ( )
# Mock out the creation of the SDRClassifier .
@ staticmethod
def _mockCreate ( * args , ** kwargs ) :
    kwargs . pop ( 'implementation' , None )
    return SDRClassifierDiff ( * args , ** kwargs )
SDRClassifierFactory . create = _mockCreate
# Run it !
runExperiment ( sys . argv [ 1 : ] )


signature:
def _abbreviate ( text , threshold ) :
    """Abbreviate the given text to threshold chars and append an ellipsis if its
 length exceeds threshold ; used for logging ;
 NOTE : the resulting text could be longer than threshold due to the ellipsis"""

implementation:
if text is not None and len ( text ) > threshold :
    text = text [ : threshold ] + "..."
return text


signature:
def __getDBNameForVersion ( cls , dbVersion ) :
    """Generates the ClientJobs database name for the given version of the
 database
 Parameters :
 dbVersion : ClientJobs database version number
 retval : the ClientJobs database name for the given DB version"""

implementation:
# DB Name prefix for the given version
prefix = cls . __getDBNamePrefixForVersion ( dbVersion )
# DB Name suffix
suffix = Configuration . get ( 'nupic.cluster.database.nameSuffix' )
# Replace dash and dot with underscore ( e . g . ' ec2 - user ' or ec2 . user will break SQL )
suffix = suffix . replace ( "-" , "_" )
suffix = suffix . replace ( "." , "_" )
# Create the name of the database for the given DB version
dbName = '%s_%s' % ( prefix , suffix )
return dbName


signature:
def get ( ) :
    """Get the instance of the ClientJobsDAO created for this process ( or
 perhaps at some point in the future , for this thread ) .
 Parameters :
 retval : instance of ClientJobsDAO"""

implementation:
# Instantiate if needed
if ClientJobsDAO . _instance is None :
    cjDAO = ClientJobsDAO ( )
    cjDAO . connect ( )
    ClientJobsDAO . _instance = cjDAO
# Return the instance to the caller
return ClientJobsDAO . _instance


signature:
def _columnNameDBToPublic ( self , dbName ) :
    """Convert a database internal column name to a public name . This
 takes something of the form word1 _ word2 _ word3 and converts it to :
 word1Word2Word3 . If the db field name starts with ' _ ' , it is stripped out
 so that the name is compatible with collections . namedtuple .
 for example : _ word1 _ word2 _ word3 = > word1Word2Word3
 Parameters :
 dbName : database internal field name
 retval : public name"""

implementation:
words = dbName . split ( '_' )
if dbName . startswith ( '_' ) :
    words = words [ 1 : ]
pubWords = [ words [ 0 ] ]
for word in words [ 1 : ] :
    pubWords . append ( word [ 0 ] . upper ( ) + word [ 1 : ] )
return '' . join ( pubWords )


signature:
def connect ( self , deleteOldVersions = False , recreate = False ) :
    """Locate the current version of the jobs DB or create a new one , and
 optionally delete old versions laying around . If desired , this method
 can be called at any time to re - create the tables from scratch , delete
 old versions of the database , etc .
 Parameters :
 deleteOldVersions : if true , delete any old versions of the DB left
 on the server
 recreate : if true , recreate the database from scratch even
 if it already exists ."""

implementation:
# Initialize tables , if needed
with ConnectionFactory . get ( ) as conn : # Initialize tables
    self . _initTables ( cursor = conn . cursor , deleteOldVersions = deleteOldVersions , recreate = recreate )
    # Save our connection id
    conn . cursor . execute ( 'SELECT CONNECTION_ID()' )
    self . _connectionID = conn . cursor . fetchall ( ) [ 0 ] [ 0 ]
    self . _logger . info ( "clientJobsConnectionID=%r" , self . _connectionID )
return


signature:
def _initTables ( self , cursor , deleteOldVersions , recreate ) :
    """Initialize tables , if needed
 Parameters :
 cursor : SQL cursor
 deleteOldVersions : if true , delete any old versions of the DB left
 on the server
 recreate : if true , recreate the database from scratch even
 if it already exists ."""

implementation:
# Delete old versions if they exist
if deleteOldVersions :
    self . _logger . info ( "Dropping old versions of client_jobs DB; called from: %r" , traceback . format_stack ( ) )
    for i in range ( self . _DB_VERSION ) :
        cursor . execute ( 'DROP DATABASE IF EXISTS %s' % ( self . __getDBNameForVersion ( i ) , ) )
# Create the database if necessary
if recreate :
    self . _logger . info ( "Dropping client_jobs DB %r; called from: %r" , self . dbName , traceback . format_stack ( ) )
    cursor . execute ( 'DROP DATABASE IF EXISTS %s' % ( self . dbName ) )
cursor . execute ( 'CREATE DATABASE IF NOT EXISTS %s' % ( self . dbName ) )
# Get the list of tables
cursor . execute ( 'SHOW TABLES IN %s' % ( self . dbName ) )
output = cursor . fetchall ( )
tableNames = [ x [ 0 ] for x in output ]
# Create the jobs table if it doesn ' t exist
# Fields that start with ' _ eng ' are intended for private use by the engine
# and should not be used by the UI
if 'jobs' not in tableNames :
    self . _logger . info ( "Creating table %r" , self . jobsTableName )
    fields = [ 'job_id                  INT UNSIGNED NOT NULL AUTO_INCREMENT' , # unique jobID
    'client                  CHAR(%d)' % ( self . CLIENT_MAX_LEN ) , # name of client ( UI , StrmMgr , etc . )
    'client_info             LONGTEXT' , # Arbitrary data defined by the client
    'client_key             varchar(255)' , # Foreign key as defined by the client .
    'cmd_line                LONGTEXT' , # command line to use to launch each worker process
    'params                  LONGTEXT' , # JSON encoded params for the job , for use by the worker processes
    'job_hash                BINARY(%d) DEFAULT NULL' % ( self . HASH_MAX_LEN ) , # unique hash of the job , provided by the client . Used for detecting
    # identical job requests from the same client when they use the
    # jobInsertUnique ( ) method .
    'status                  VARCHAR(16) DEFAULT "notStarted"' , # One of the STATUS _ XXX enumerated value strings
    'completion_reason       VARCHAR(16)' , # One of the CMPL _ REASON _ XXX enumerated value strings .
    # NOTE : This is the job completion reason according to the hadoop
    # job - tracker . A success here does not necessarily mean the
    # workers were " happy " with the job . To see if the workers
    # failed , check the worker _ completion _ reason
    'completion_msg          LONGTEXT' , # Why this job completed , according to job - tracker
    'worker_completion_reason   VARCHAR(16) DEFAULT "%s"' % self . CMPL_REASON_SUCCESS , # One of the CMPL _ REASON _ XXX enumerated value strings . This is
    # may be changed to CMPL _ REASON _ ERROR if any workers encounter
    # an error while running the job .
    'worker_completion_msg   LONGTEXT' , # Why this job completed , according to workers . If
    # worker _ completion _ reason is set to CMPL _ REASON _ ERROR , this will
    # contain the error information .
    'cancel                  BOOLEAN DEFAULT FALSE' , # set by UI , polled by engine
    'start_time              DATETIME DEFAULT NULL' , # When job started
    'end_time                DATETIME DEFAULT NULL' , # When job ended
    'results                 LONGTEXT' , # JSON dict with general information about the results of the job ,
    # including the ID and value of the best model
    # TODO : different semantics for results field of ProductionJob
    '_eng_job_type           VARCHAR(32)' , # String used to specify the type of job that this is . Current
    # choices are hypersearch , production worker , or stream worker
    'minimum_workers         INT UNSIGNED DEFAULT 0' , # min number of desired workers at a time . If 0 , no workers will be
    # allocated in a crunch
    'maximum_workers         INT UNSIGNED DEFAULT 0' , # max number of desired workers at a time . If 0 , then use as many
    # as practical given load on the cluster .
    'priority                 INT DEFAULT %d' % self . DEFAULT_JOB_PRIORITY , # job scheduling priority ; 0 is the default priority (
    # ClientJobsDAO . DEFAULT _ JOB _ PRIORITY ) ; positive values are higher
    # priority ( up to ClientJobsDAO . MAX _ JOB _ PRIORITY ) , and negative
    # values are lower priority ( down to ClientJobsDAO . MIN _ JOB _ PRIORITY )
    '_eng_allocate_new_workers    BOOLEAN DEFAULT TRUE' , # Should the scheduling algorithm allocate new workers to this job ?
    # If a specialized worker willingly gives up control , we set this
    # field to FALSE to avoid allocating new workers .
    '_eng_untended_dead_workers   BOOLEAN DEFAULT FALSE' , # If a specialized worker fails or is killed by the scheduler , we
    # set this feild to TRUE to indicate that the worker is dead
    'num_failed_workers           INT UNSIGNED DEFAULT 0' , # The number of failed specialized workers for this job . If the
    # number of failures is > = max . failed . attempts , we mark the job
    # as failed
    'last_failed_worker_error_msg  LONGTEXT' , # Error message of the most recent specialized failed worker
    '_eng_cleaning_status          VARCHAR(16) DEFAULT "%s"' % self . CLEAN_NOT_DONE , # Has the job been garbage collected , this includes removing
    # unneeded # model output caches , s3 checkpoints .
    'gen_base_description    LONGTEXT' , # The contents of the generated description . py file from hypersearch
    # requests . This is generated by the Hypersearch workers and stored
    # here for reference , debugging , and development purposes .
    'gen_permutations        LONGTEXT' , # The contents of the generated permutations . py file from
    # hypersearch requests . This is generated by the Hypersearch workers
    # and stored here for reference , debugging , and development
    # purposes .
    '_eng_last_update_time   DATETIME DEFAULT NULL' , # time stamp of last update , used for detecting stalled jobs
    '_eng_cjm_conn_id        INT UNSIGNED' , # ID of the CJM starting up this job
    '_eng_worker_state       LONGTEXT' , # JSON encoded state of the hypersearch in progress , for private
    # use by the Hypersearch workers
    '_eng_status             LONGTEXT' , # String used for status messages sent from the engine for
    # informative purposes only . Usually printed periodically by
    # clients watching a job progress .
    '_eng_model_milestones   LONGTEXT' , # JSon encoded object with information about global model milestone
    # results
    'PRIMARY KEY (job_id)' , 'UNIQUE INDEX (client, job_hash)' , 'INDEX (status)' , 'INDEX (client_key)' ]
    options = [ 'AUTO_INCREMENT=1000' , ]
    query = 'CREATE TABLE IF NOT EXISTS %s (%s) %s' % ( self . jobsTableName , ',' . join ( fields ) , ',' . join ( options ) )
    cursor . execute ( query )
# Create the models table if it doesn ' t exist
# Fields that start with ' _ eng ' are intended for private use by the engine
# and should not be used by the UI
if 'models' not in tableNames :
    self . _logger . info ( "Creating table %r" , self . modelsTableName )
    fields = [ 'model_id                BIGINT UNSIGNED NOT NULL AUTO_INCREMENT' , # globally unique model ID
    'job_id                  INT UNSIGNED NOT NULL' , # jobID
    'params                  LONGTEXT NOT NULL' , # JSON encoded params for the model
    'status                  VARCHAR(16) DEFAULT "notStarted"' , # One of the STATUS _ XXX enumerated value strings
    'completion_reason       VARCHAR(16)' , # One of the CMPL _ REASON _ XXX enumerated value strings
    'completion_msg          LONGTEXT' , # Why this job completed
    'results                 LONGTEXT DEFAULT NULL' , # JSON encoded structure containing metrics produced by the model
    'optimized_metric        FLOAT ' , # Value of the particular metric we are optimizing in hypersearch
    'update_counter          INT UNSIGNED DEFAULT 0' , # incremented by engine every time the results is updated
    'num_records             INT UNSIGNED DEFAULT 0' , # number of records processed so far
    'start_time              DATETIME DEFAULT NULL' , # When this model started being evaluated
    'end_time                DATETIME DEFAULT NULL' , # When this model completed
    'cpu_time                FLOAT DEFAULT 0' , # How much actual CPU time was spent on this model , in seconds . This
    # excludes time the process spent sleeping , or otherwise not
    # actually executing code .
    'model_checkpoint_id     LONGTEXT' , # Checkpoint identifier for this model ( after it has been saved )
    'gen_description         LONGTEXT' , # The contents of the generated description . py file from hypersearch
    # requests . This is generated by the Hypersearch workers and stored
    # here for reference , debugging , and development purposes .
    '_eng_params_hash        BINARY(%d) DEFAULT NULL' % ( self . HASH_MAX_LEN ) , # MD5 hash of the params
    '_eng_particle_hash      BINARY(%d) DEFAULT NULL' % ( self . HASH_MAX_LEN ) , # MD5 hash of the particle info for PSO algorithm
    '_eng_last_update_time   DATETIME DEFAULT NULL' , # time stamp of last update , used for detecting stalled workers
    '_eng_task_tracker_id    TINYBLOB' , # Hadoop Task Tracker ID
    '_eng_worker_id          TINYBLOB' , # Hadoop Map Task ID
    '_eng_attempt_id         TINYBLOB' , # Hadoop Map task attempt ID
    '_eng_worker_conn_id     INT DEFAULT 0' , # database client connection ID of the worker that is running this
    # model
    '_eng_milestones         LONGTEXT' , # A JSON encoded list of metric values for the model at each
    # milestone point
    '_eng_stop               VARCHAR(16) DEFAULT NULL' , # One of the STOP _ REASON _ XXX enumerated value strings . Set either by
    # the swarm terminator of either the current , or another
    # Hypersearch worker .
    '_eng_matured            BOOLEAN DEFAULT FALSE' , # Set by the model maturity - checker when it decides that this model
    # has " matured " . This means that it has reached the point of
    # not getting better results with more data .
    'PRIMARY KEY (model_id)' , 'UNIQUE INDEX (job_id, _eng_params_hash)' , 'UNIQUE INDEX (job_id, _eng_particle_hash)' , ]
    options = [ 'AUTO_INCREMENT=1000' , ]
    query = 'CREATE TABLE IF NOT EXISTS %s (%s) %s' % ( self . modelsTableName , ',' . join ( fields ) , ',' . join ( options ) )
    cursor . execute ( query )
# Get the field names for each table
cursor . execute ( 'DESCRIBE %s' % ( self . jobsTableName ) )
fields = cursor . fetchall ( )
self . _jobs . dbFieldNames = [ str ( field [ 0 ] ) for field in fields ]
cursor . execute ( 'DESCRIBE %s' % ( self . modelsTableName ) )
fields = cursor . fetchall ( )
self . _models . dbFieldNames = [ str ( field [ 0 ] ) for field in fields ]
# Generate the public names
self . _jobs . publicFieldNames = [ self . _columnNameDBToPublic ( x ) for x in self . _jobs . dbFieldNames ]
self . _models . publicFieldNames = [ self . _columnNameDBToPublic ( x ) for x in self . _models . dbFieldNames ]
# Generate the name conversion dicts
self . _jobs . pubToDBNameDict = dict ( zip ( self . _jobs . publicFieldNames , self . _jobs . dbFieldNames ) )
self . _jobs . dbToPubNameDict = dict ( zip ( self . _jobs . dbFieldNames , self . _jobs . publicFieldNames ) )
self . _models . pubToDBNameDict = dict ( zip ( self . _models . publicFieldNames , self . _models . dbFieldNames ) )
self . _models . dbToPubNameDict = dict ( zip ( self . _models . dbFieldNames , self . _models . publicFieldNames ) )
# Generate the dynamic namedtuple classes we use
self . _models . modelInfoNamedTuple = collections . namedtuple ( '_modelInfoNamedTuple' , self . _models . publicFieldNames )
self . _jobs . jobInfoNamedTuple = collections . namedtuple ( '_jobInfoNamedTuple' , self . _jobs . publicFieldNames )
return


signature:
def _getMatchingRowsNoRetries ( self , tableInfo , conn , fieldsToMatch , selectFieldNames , maxRows = None ) :
    """Return a sequence of matching rows with the requested field values from
 a table or empty sequence if nothing matched .
 tableInfo : Table information : a ClientJobsDAO . _ TableInfoBase instance
 conn : Owned connection acquired from ConnectionFactory . get ( )
 fieldsToMatch : Dictionary of internal fieldName / value mappings that
 identify the desired rows . If a value is an instance of
 ClientJobsDAO . _ SEQUENCE _ TYPES ( list / set / tuple ) , then the
 operator ' IN ' will be used in the corresponding SQL
 predicate ; if the value is bool : " IS TRUE / FALSE " ; if the
 value is None : " IS NULL " ; ' = ' will be used for all other
 cases .
 selectFieldNames :
 list of fields to return , using internal field names
 maxRows : maximum number of rows to return ; unlimited if maxRows
 is None
 retval : A sequence of matching rows , each row consisting of field
 values in the order of the requested field names . Empty
 sequence is returned when not match exists ."""

implementation:
assert fieldsToMatch , repr ( fieldsToMatch )
assert all ( k in tableInfo . dbFieldNames for k in fieldsToMatch . iterkeys ( ) ) , repr ( fieldsToMatch )
assert selectFieldNames , repr ( selectFieldNames )
assert all ( f in tableInfo . dbFieldNames for f in selectFieldNames ) , repr ( selectFieldNames )
# NOTE : make sure match expressions and values are in the same order
matchPairs = fieldsToMatch . items ( )
matchExpressionGen = ( p [ 0 ] + ( ' IS ' + { True : 'TRUE' , False : 'FALSE' } [ p [ 1 ] ] if isinstance ( p [ 1 ] , bool ) else ' IS NULL' if p [ 1 ] is None else ' IN %s' if isinstance ( p [ 1 ] , self . _SEQUENCE_TYPES ) else '=%s' ) for p in matchPairs )
matchFieldValues = [ p [ 1 ] for p in matchPairs if ( not isinstance ( p [ 1 ] , ( bool ) ) and p [ 1 ] is not None ) ]
query = 'SELECT %s FROM %s WHERE (%s)' % ( ',' . join ( selectFieldNames ) , tableInfo . tableName , ' AND ' . join ( matchExpressionGen ) )
sqlParams = matchFieldValues
if maxRows is not None :
    query += ' LIMIT %s'
    sqlParams . append ( maxRows )
conn . cursor . execute ( query , sqlParams )
rows = conn . cursor . fetchall ( )
if rows :
    assert maxRows is None or len ( rows ) <= maxRows , "%d !<= %d" % ( len ( rows ) , maxRows )
    assert len ( rows [ 0 ] ) == len ( selectFieldNames ) , "%d != %d" % ( len ( rows [ 0 ] ) , len ( selectFieldNames ) )
else :
    rows = tuple ( )
return rows


signature:
def _getMatchingRowsWithRetries ( self , tableInfo , fieldsToMatch , selectFieldNames , maxRows = None ) :
    """Like _ getMatchingRowsNoRetries ( ) , but with retries on transient MySQL
 failures"""

implementation:
with ConnectionFactory . get ( ) as conn :
    return self . _getMatchingRowsNoRetries ( tableInfo , conn , fieldsToMatch , selectFieldNames , maxRows )


signature:
def _getOneMatchingRowNoRetries ( self , tableInfo , conn , fieldsToMatch , selectFieldNames ) :
    """Return a single matching row with the requested field values from the
 the requested table or None if nothing matched .
 tableInfo : Table information : a ClientJobsDAO . _ TableInfoBase instance
 conn : Owned connection acquired from ConnectionFactory . get ( )
 fieldsToMatch : Dictionary of internal fieldName / value mappings that
 identify the desired rows . If a value is an instance of
 ClientJobsDAO . _ SEQUENCE _ TYPES ( list / set / tuple ) , then the
 operator ' IN ' will be used in the corresponding SQL
 predicate ; if the value is bool : " IS TRUE / FALSE " ; if the
 value is None : " IS NULL " ; ' = ' will be used for all other
 cases .
 selectFieldNames :
 list of fields to return , using internal field names
 retval : A sequence of field values of the matching row in the order
 of the given field names ; or None if there was no match ."""

implementation:
rows = self . _getMatchingRowsNoRetries ( tableInfo , conn , fieldsToMatch , selectFieldNames , maxRows = 1 )
if rows :
    assert len ( rows ) == 1 , repr ( len ( rows ) )
    result = rows [ 0 ]
else :
    result = None
return result


signature:
def _getOneMatchingRowWithRetries ( self , tableInfo , fieldsToMatch , selectFieldNames ) :
    """Like _ getOneMatchingRowNoRetries ( ) , but with retries on transient MySQL
 failures"""

implementation:
with ConnectionFactory . get ( ) as conn :
    return self . _getOneMatchingRowNoRetries ( tableInfo , conn , fieldsToMatch , selectFieldNames )


signature:
def _insertOrGetUniqueJobNoRetries ( self , conn , client , cmdLine , jobHash , clientInfo , clientKey , params , minimumWorkers , maximumWorkers , jobType , priority , alreadyRunning ) :
    """Attempt to insert a row with the given parameters into the jobs table .
 Return jobID of the inserted row , or of an existing row with matching
 client / jobHash key .
 The combination of client and jobHash are expected to be unique ( enforced
 by a unique index on the two columns ) .
 NOTE : It ' s possibe that this or another process ( on this or another machine )
 already inserted a row with matching client / jobHash key ( e . g . ,
 StreamMgr ) . This may also happen undetected by this function due to a
 partially - successful insert operation ( e . g . , row inserted , but then
 connection was lost while reading response ) followed by retries either of
 this function or in SteadyDB module .
 Parameters :
 conn : Owned connection acquired from ConnectionFactory . get ( )
 client : Name of the client submitting the job
 cmdLine : Command line to use to launch each worker process ; must be
 a non - empty string
 jobHash : unique hash of this job . The caller must insure that this ,
 together with client , uniquely identifies this job request
 for the purposes of detecting duplicates .
 clientInfo : JSON encoded dict of client specific information .
 clientKey : Foreign key .
 params : JSON encoded dict of the parameters for the job . This
 can be fetched out of the database by the worker processes
 based on the jobID .
 minimumWorkers : minimum number of workers design at a time .
 maximumWorkers : maximum number of workers desired at a time .
 priority : Job scheduling priority ; 0 is the default priority (
 ClientJobsDAO . DEFAULT _ JOB _ PRIORITY ) ; positive values are
 higher priority ( up to ClientJobsDAO . MAX _ JOB _ PRIORITY ) ,
 and negative values are lower priority ( down to
 ClientJobsDAO . MIN _ JOB _ PRIORITY ) . Higher - priority jobs will
 be scheduled to run at the expense of the lower - priority
 jobs , and higher - priority job tasks will preempt those
 with lower priority if there is inadequate supply of
 scheduling slots . Excess lower priority job tasks will
 starve as long as slot demand exceeds supply . Most jobs
 should be scheduled with DEFAULT _ JOB _ PRIORITY . System jobs
 that must run at all cost , such as Multi - Model - Master ,
 should be scheduled with MAX _ JOB _ PRIORITY .
 alreadyRunning : Used for unit test purposes only . This inserts the job
 in the running state . It is used when running a worker
 in standalone mode without hadoop - it gives it a job
 record to work with .
 retval : jobID of the inserted jobs row , or of an existing jobs row
 with matching client / jobHash key"""

implementation:
assert len ( client ) <= self . CLIENT_MAX_LEN , "client too long:" + repr ( client )
assert cmdLine , "Unexpected empty or None command-line: " + repr ( cmdLine )
assert len ( jobHash ) == self . HASH_MAX_LEN , "wrong hash len=%d" % len ( jobHash )
# Initial status
if alreadyRunning : # STATUS _ TESTMODE , so that scheduler won ' t pick it up ( for in - proc tests )
    initStatus = self . STATUS_TESTMODE
else :
    initStatus = self . STATUS_NOTSTARTED
# Create a new job entry
query = 'INSERT IGNORE INTO %s (status, client, client_info, client_key,' 'cmd_line, params, job_hash, _eng_last_update_time, ' 'minimum_workers, maximum_workers, priority, _eng_job_type) ' ' VALUES (%%s, %%s, %%s, %%s, %%s, %%s, %%s, ' '         UTC_TIMESTAMP(), %%s, %%s, %%s, %%s) ' % ( self . jobsTableName , )
sqlParams = ( initStatus , client , clientInfo , clientKey , cmdLine , params , jobHash , minimumWorkers , maximumWorkers , priority , jobType )
numRowsInserted = conn . cursor . execute ( query , sqlParams )
jobID = 0
if numRowsInserted == 1 : # Get the chosen job id
# NOTE : LAST _ INSERT _ ID ( ) returns 0 after intermittent connection failure
    conn . cursor . execute ( 'SELECT LAST_INSERT_ID()' )
    jobID = conn . cursor . fetchall ( ) [ 0 ] [ 0 ]
    if jobID == 0 :
        self . _logger . warn ( '_insertOrGetUniqueJobNoRetries: SELECT LAST_INSERT_ID() returned 0; ' 'likely due to reconnection in SteadyDB following INSERT. ' 'jobType=%r; client=%r; clientInfo=%r; clientKey=%s; jobHash=%r; ' 'cmdLine=%r' , jobType , client , _abbreviate ( clientInfo , 32 ) , clientKey , jobHash , cmdLine )
else : # Assumption : nothing was inserted because this is a retry and the row
# with this client / hash already exists from our prior
# partially - successful attempt ; or row with matching client / jobHash was
# inserted already by some process on some machine .
    assert numRowsInserted == 0 , repr ( numRowsInserted )
if jobID == 0 : # Recover from intermittent failure in a partially - successful attempt ;
# or row with matching client / jobHash was already in table
    row = self . _getOneMatchingRowNoRetries ( self . _jobs , conn , dict ( client = client , job_hash = jobHash ) , [ 'job_id' ] )
    assert row is not None
    assert len ( row ) == 1 , 'Unexpected num fields: ' + repr ( len ( row ) )
    jobID = row [ 0 ]
# If asked to enter the job in the running state , set the connection id
# and start time as well
if alreadyRunning :
    query = 'UPDATE %s SET _eng_cjm_conn_id=%%s, ' '              start_time=UTC_TIMESTAMP(), ' '              _eng_last_update_time=UTC_TIMESTAMP() ' '          WHERE job_id=%%s' % ( self . jobsTableName , )
    conn . cursor . execute ( query , ( self . _connectionID , jobID ) )
return jobID


signature:
def _resumeJobNoRetries ( self , conn , jobID , alreadyRunning ) :
    """Resumes processing of an existing job that is presently in the
 STATUS _ COMPLETED state .
 NOTE : this is primarily for resuming suspended Production and Stream Jobs ; DO
 NOT use it on Hypersearch jobs .
 This prepares an existing job entry to resume processing . The CJM is always
 periodically sweeping the jobs table and when it finds a job that is ready
 to run , it will proceed to start it up on Hadoop .
 Parameters :
 conn : Owned connection acquired from ConnectionFactory . get ( )
 jobID : jobID of the job to resume
 alreadyRunning : Used for unit test purposes only . This inserts the job
 in the running state . It is used when running a worker
 in standalone mode without hadoop .
 raises : Throws a RuntimeError if no rows are affected . This could
 either be because :
 1 ) Because there was not matching jobID
 2 ) or if the status of the job was not STATUS _ COMPLETED .
 retval : nothing"""

implementation:
# Initial status
if alreadyRunning : # Use STATUS _ TESTMODE so scheduler will leave our row alone
    initStatus = self . STATUS_TESTMODE
else :
    initStatus = self . STATUS_NOTSTARTED
# NOTE : some of our clients ( e . g . , StreamMgr ) may call us ( directly or
# indirectly ) for the same job from different processes ( even different
# machines ) , so we should be prepared for the update to fail ; same holds
# if the UPDATE succeeds , but connection fails while reading result
assignments = [ 'status=%s' , 'completion_reason=DEFAULT' , 'completion_msg=DEFAULT' , 'worker_completion_reason=DEFAULT' , 'worker_completion_msg=DEFAULT' , 'end_time=DEFAULT' , 'cancel=DEFAULT' , '_eng_last_update_time=UTC_TIMESTAMP()' , '_eng_allocate_new_workers=DEFAULT' , '_eng_untended_dead_workers=DEFAULT' , 'num_failed_workers=DEFAULT' , 'last_failed_worker_error_msg=DEFAULT' , '_eng_cleaning_status=DEFAULT' , ]
assignmentValues = [ initStatus ]
if alreadyRunning :
    assignments += [ '_eng_cjm_conn_id=%s' , 'start_time=UTC_TIMESTAMP()' , '_eng_last_update_time=UTC_TIMESTAMP()' ]
    assignmentValues . append ( self . _connectionID )
else :
    assignments += [ '_eng_cjm_conn_id=DEFAULT' , 'start_time=DEFAULT' ]
assignments = ', ' . join ( assignments )
query = 'UPDATE %s SET %s ' '          WHERE job_id=%%s AND status=%%s' % ( self . jobsTableName , assignments )
sqlParams = assignmentValues + [ jobID , self . STATUS_COMPLETED ]
numRowsAffected = conn . cursor . execute ( query , sqlParams )
assert numRowsAffected <= 1 , repr ( numRowsAffected )
if numRowsAffected == 0 :
    self . _logger . info ( "_resumeJobNoRetries: Redundant job-resume UPDATE: job was not " "suspended or was resumed by another process or operation was retried " "after connection failure; jobID=%s" , jobID )
return


signature:
def jobResume ( self , jobID , alreadyRunning = False ) :
    """Resumes processing of an existing job that is presently in the
 STATUS _ COMPLETED state .
 NOTE : this is primarily for resuming suspended Production Jobs ; DO NOT use
 it on Hypersearch jobs .
 NOTE : The job MUST be in the STATUS _ COMPLETED state at the time of this
 call , otherwise an exception will be raised .
 This prepares an existing job entry to resume processing . The CJM is always
 periodically sweeping the jobs table and when it finds a job that is ready
 to run , will proceed to start it up on Hadoop .
 Parameters :
 job : jobID of the job to resume
 alreadyRunning : Used for unit test purposes only . This inserts the job
 in the running state . It is used when running a worker
 in standalone mode without hadoop .
 raises : Throws a RuntimeError if no rows are affected . This could
 either be because :
 1 ) Because there was not matching jobID
 2 ) or if the status of the job was not STATUS _ COMPLETED .
 retval : nothing"""

implementation:
row = self . jobGetFields ( jobID , [ 'status' ] )
( jobStatus , ) = row
if jobStatus != self . STATUS_COMPLETED :
    raise RuntimeError ( ( "Failed to resume job: job was not suspended; " "jobID=%s; job status=%r" ) % ( jobID , jobStatus ) )
# NOTE : on MySQL failures , we need to retry ConnectionFactory . get ( ) as well
# in order to recover from lost connections
@ g_retrySQL
def resumeWithRetries ( ) :
    with ConnectionFactory . get ( ) as conn :
        self . _resumeJobNoRetries ( conn , jobID , alreadyRunning )
resumeWithRetries ( )
return


signature:
def jobInsert ( self , client , cmdLine , clientInfo = '' , clientKey = '' , params = '' , alreadyRunning = False , minimumWorkers = 0 , maximumWorkers = 0 , jobType = '' , priority = DEFAULT_JOB_PRIORITY ) :
    """Add an entry to the jobs table for a new job request . This is called by
 clients that wish to startup a new job , like a Hypersearch , stream job , or
 specific model evaluation from the engine .
 This puts a new entry into the jobs table . The CJM is always periodically
 sweeping the jobs table and when it finds a new job , will proceed to start it
 up on Hadoop .
 Parameters :
 client : Name of the client submitting the job
 cmdLine : Command line to use to launch each worker process ; must be
 a non - empty string
 clientInfo : JSON encoded dict of client specific information .
 clientKey : Foreign key .
 params : JSON encoded dict of the parameters for the job . This
 can be fetched out of the database by the worker processes
 based on the jobID .
 alreadyRunning : Used for unit test purposes only . This inserts the job
 in the running state . It is used when running a worker
 in standalone mode without hadoop - it gives it a job
 record to work with .
 minimumWorkers : minimum number of workers design at a time .
 maximumWorkers : maximum number of workers desired at a time .
 jobType : The type of job that this is . This should be one of the
 JOB _ TYPE _ XXXX enums . This is needed to allow a standard
 way of recognizing a job ' s function and capabilities .
 priority : Job scheduling priority ; 0 is the default priority (
 ClientJobsDAO . DEFAULT _ JOB _ PRIORITY ) ; positive values are
 higher priority ( up to ClientJobsDAO . MAX _ JOB _ PRIORITY ) ,
 and negative values are lower priority ( down to
 ClientJobsDAO . MIN _ JOB _ PRIORITY ) . Higher - priority jobs will
 be scheduled to run at the expense of the lower - priority
 jobs , and higher - priority job tasks will preempt those
 with lower priority if there is inadequate supply of
 scheduling slots . Excess lower priority job tasks will
 starve as long as slot demand exceeds supply . Most jobs
 should be scheduled with DEFAULT _ JOB _ PRIORITY . System jobs
 that must run at all cost , such as Multi - Model - Master ,
 should be scheduled with MAX _ JOB _ PRIORITY .
 retval : jobID - unique ID assigned to this job"""

implementation:
jobHash = self . _normalizeHash ( uuid . uuid1 ( ) . bytes )
@ g_retrySQL
def insertWithRetries ( ) :
    with ConnectionFactory . get ( ) as conn :
        return self . _insertOrGetUniqueJobNoRetries ( conn , client = client , cmdLine = cmdLine , jobHash = jobHash , clientInfo = clientInfo , clientKey = clientKey , params = params , minimumWorkers = minimumWorkers , maximumWorkers = maximumWorkers , jobType = jobType , priority = priority , alreadyRunning = alreadyRunning )
try :
    jobID = insertWithRetries ( )
except :
    self . _logger . exception ( 'jobInsert FAILED: jobType=%r; client=%r; clientInfo=%r; clientKey=%r;' 'jobHash=%r; cmdLine=%r' , jobType , client , _abbreviate ( clientInfo , 48 ) , clientKey , jobHash , cmdLine )
    raise
else :
    self . _logger . info ( 'jobInsert: returning jobID=%s. jobType=%r; client=%r; clientInfo=%r; ' 'clientKey=%r; jobHash=%r; cmdLine=%r' , jobID , jobType , client , _abbreviate ( clientInfo , 48 ) , clientKey , jobHash , cmdLine )
return jobID


signature:
def jobInsertUnique ( self , client , cmdLine , jobHash , clientInfo = '' , clientKey = '' , params = '' , minimumWorkers = 0 , maximumWorkers = 0 , jobType = '' , priority = DEFAULT_JOB_PRIORITY ) :
    """Add an entry to the jobs table for a new job request , but only if the
 same job , by the same client is not already running . If the job is already
 running , or queued up to run , this call does nothing . If the job does not
 exist in the jobs table or has completed , it will be inserted and / or started
 up again .
 This method is called by clients , like StreamMgr , that wish to only start up
 a job if it hasn ' t already been started up .
 Parameters :
 client : Name of the client submitting the job
 cmdLine : Command line to use to launch each worker process ; must be
 a non - empty string
 jobHash : unique hash of this job . The client must insure that this
 uniquely identifies this job request for the purposes
 of detecting duplicates .
 clientInfo : JSON encoded dict of client specific information .
 clientKey : Foreign key .
 params : JSON encoded dict of the parameters for the job . This
 can be fetched out of the database by the worker processes
 based on the jobID .
 minimumWorkers : minimum number of workers design at a time .
 maximumWorkers : maximum number of workers desired at a time .
 jobType : The type of job that this is . This should be one of the
 JOB _ TYPE _ XXXX enums . This is needed to allow a standard
 way of recognizing a job ' s function and capabilities .
 priority : Job scheduling priority ; 0 is the default priority (
 ClientJobsDAO . DEFAULT _ JOB _ PRIORITY ) ; positive values are
 higher priority ( up to ClientJobsDAO . MAX _ JOB _ PRIORITY ) ,
 and negative values are lower priority ( down to
 ClientJobsDAO . MIN _ JOB _ PRIORITY ) . Higher - priority jobs will
 be scheduled to run at the expense of the lower - priority
 jobs , and higher - priority job tasks will preempt those
 with lower priority if there is inadequate supply of
 scheduling slots . Excess lower priority job tasks will
 starve as long as slot demand exceeds supply . Most jobs
 should be scheduled with DEFAULT _ JOB _ PRIORITY . System jobs
 that must run at all cost , such as Multi - Model - Master ,
 should be scheduled with MAX _ JOB _ PRIORITY .
 retval : jobID of the newly inserted or existing job ."""

implementation:
assert cmdLine , "Unexpected empty or None command-line: " + repr ( cmdLine )
@ g_retrySQL
def insertUniqueWithRetries ( ) :
    jobHashValue = self . _normalizeHash ( jobHash )
    jobID = None
    with ConnectionFactory . get ( ) as conn :
        row = self . _getOneMatchingRowNoRetries ( self . _jobs , conn , dict ( client = client , job_hash = jobHashValue ) , [ 'job_id' , 'status' ] )
        if row is not None :
            ( jobID , status ) = row
            if status == self . STATUS_COMPLETED : # Restart existing job that had completed
                query = 'UPDATE %s SET client_info=%%s, ' '              client_key=%%s, ' '              cmd_line=%%s, ' '              params=%%s, ' '              minimum_workers=%%s, ' '              maximum_workers=%%s, ' '              priority=%%s, ' '              _eng_job_type=%%s ' '          WHERE (job_id=%%s AND status=%%s)' % ( self . jobsTableName , )
                sqlParams = ( clientInfo , clientKey , cmdLine , params , minimumWorkers , maximumWorkers , priority , jobType , jobID , self . STATUS_COMPLETED )
                numRowsUpdated = conn . cursor . execute ( query , sqlParams )
                assert numRowsUpdated <= 1 , repr ( numRowsUpdated )
                if numRowsUpdated == 0 :
                    self . _logger . info ( "jobInsertUnique: Redundant job-reuse UPDATE: job restarted by " "another process, values were unchanged, or operation was " "retried after connection failure; jobID=%s" , jobID )
                # Restart the job , unless another process beats us to it
                self . _resumeJobNoRetries ( conn , jobID , alreadyRunning = False )
        else : # There was no job row with matching client / jobHash , so insert one
            jobID = self . _insertOrGetUniqueJobNoRetries ( conn , client = client , cmdLine = cmdLine , jobHash = jobHashValue , clientInfo = clientInfo , clientKey = clientKey , params = params , minimumWorkers = minimumWorkers , maximumWorkers = maximumWorkers , jobType = jobType , priority = priority , alreadyRunning = False )
        return jobID
try :
    jobID = insertUniqueWithRetries ( )
except :
    self . _logger . exception ( 'jobInsertUnique FAILED: jobType=%r; client=%r; ' 'clientInfo=%r; clientKey=%r; jobHash=%r; cmdLine=%r' , jobType , client , _abbreviate ( clientInfo , 48 ) , clientKey , jobHash , cmdLine )
    raise
else :
    self . _logger . info ( 'jobInsertUnique: returning jobID=%s. jobType=%r; client=%r; ' 'clientInfo=%r; clientKey=%r; jobHash=%r; cmdLine=%r' , jobID , jobType , client , _abbreviate ( clientInfo , 48 ) , clientKey , jobHash , cmdLine )
return jobID


signature:
def _startJobWithRetries ( self , jobID ) :
    """Place the given job in STATUS _ RUNNING mode ; the job is expected to be
 STATUS _ NOTSTARTED .
 NOTE : this function was factored out of jobStartNext because it ' s also
 needed for testing ( e . g . , test _ client _ jobs _ dao . py )"""

implementation:
with ConnectionFactory . get ( ) as conn :
    query = 'UPDATE %s SET status=%%s, ' '            _eng_cjm_conn_id=%%s, ' '            start_time=UTC_TIMESTAMP(), ' '            _eng_last_update_time=UTC_TIMESTAMP() ' '          WHERE (job_id=%%s AND status=%%s)' % ( self . jobsTableName , )
    sqlParams = [ self . STATUS_RUNNING , self . _connectionID , jobID , self . STATUS_NOTSTARTED ]
    numRowsUpdated = conn . cursor . execute ( query , sqlParams )
    if numRowsUpdated != 1 :
        self . _logger . warn ( 'jobStartNext: numRowsUpdated=%r instead of 1; ' 'likely side-effect of transient connection ' 'failure' , numRowsUpdated )
return


signature:
def jobStartNext ( self ) :
    """For use only by Nupic Scheduler ( also known as ClientJobManager ) Look
 through the jobs table and see if any new job requests have been
 queued up . If so , pick one and mark it as starting up and create the
 model table to hold the results
 Parameters :
 retval : jobID of the job we are starting up , if found ; None if not found"""

implementation:
# NOTE : cursor . execute ( ' SELECT @ update _ id ' ) trick is unreliable : if a
# connection loss occurs during cursor . execute , then the server - cached
# information is lost , and we cannot get the updated job ID ; so , we use
# this select instead
row = self . _getOneMatchingRowWithRetries ( self . _jobs , dict ( status = self . STATUS_NOTSTARTED ) , [ 'job_id' ] )
if row is None :
    return None
( jobID , ) = row
self . _startJobWithRetries ( jobID )
return jobID


signature:
def jobReactivateRunningJobs ( self ) :
    """Look through the jobs table and reactivate all that are already in the
 running state by setting their _ eng _ allocate _ new _ workers fields to True ;
 used by Nupic Scheduler as part of its failure - recovery procedure ."""

implementation:
# Get a database connection and cursor
with ConnectionFactory . get ( ) as conn :
    query = 'UPDATE %s SET _eng_cjm_conn_id=%%s, ' '              _eng_allocate_new_workers=TRUE ' '    WHERE status=%%s ' % ( self . jobsTableName , )
    conn . cursor . execute ( query , [ self . _connectionID , self . STATUS_RUNNING ] )
return


signature:
def jobGetDemand ( self , ) :
    """Look through the jobs table and get the demand - minimum and maximum
 number of workers requested , if new workers are to be allocated , if there
 are any untended dead workers , for all running jobs .
 Parameters :
 retval : list of ClientJobsDAO . _ jobs . jobDemandNamedTuple nametuples
 containing the demand - min and max workers ,
 allocate _ new _ workers , untended _ dead _ workers , num _ failed _ workers
 for each running ( STATUS _ RUNNING ) job . Empty list when there
 isn ' t any demand ."""

implementation:
rows = self . _getMatchingRowsWithRetries ( self . _jobs , dict ( status = self . STATUS_RUNNING ) , [ self . _jobs . pubToDBNameDict [ f ] for f in self . _jobs . jobDemandNamedTuple . _fields ] )
return [ self . _jobs . jobDemandNamedTuple . _make ( r ) for r in rows ]


signature:
def jobCancelAllRunningJobs ( self ) :
    """Set cancel field of all currently - running jobs to true ."""

implementation:
# Get a database connection and cursor
with ConnectionFactory . get ( ) as conn :
    query = 'UPDATE %s SET cancel=TRUE WHERE status<>%%s ' % ( self . jobsTableName , )
    conn . cursor . execute ( query , [ self . STATUS_COMPLETED ] )
return


signature:
def jobCountCancellingJobs ( self , ) :
    """Look through the jobs table and count the running jobs whose
 cancel field is true .
 Parameters :
 retval : A count of running jobs with the cancel field set to true ."""

implementation:
with ConnectionFactory . get ( ) as conn :
    query = 'SELECT COUNT(job_id) ' 'FROM %s ' 'WHERE (status<>%%s AND cancel is TRUE)' % ( self . jobsTableName , )
    conn . cursor . execute ( query , [ self . STATUS_COMPLETED ] )
    rows = conn . cursor . fetchall ( )
return rows [ 0 ] [ 0 ]


signature:
def jobGetCancellingJobs ( self , ) :
    """Look through the jobs table and get the list of running jobs whose
 cancel field is true .
 Parameters :
 retval : A ( possibly empty ) sequence of running job IDs with cancel field
 set to true"""

implementation:
with ConnectionFactory . get ( ) as conn :
    query = 'SELECT job_id ' 'FROM %s ' 'WHERE (status<>%%s AND cancel is TRUE)' % ( self . jobsTableName , )
    conn . cursor . execute ( query , [ self . STATUS_COMPLETED ] )
    rows = conn . cursor . fetchall ( )
return tuple ( r [ 0 ] for r in rows )


signature:
def partitionAtIntervals ( data , intervals ) :
    """Generator to allow iterating slices at dynamic intervals
 Parameters :
 data : Any data structure that supports slicing ( i . e . list or tuple )
 * intervals : Iterable of intervals . The sum of intervals should be less
 than , or equal to the length of data ."""

implementation:
assert sum ( intervals ) <= len ( data )
start = 0
for interval in intervals :
    end = start + interval
    yield data [ start : end ]
    start = end
raise StopIteration


signature:
def _combineResults ( result , * namedTuples ) :
    """Return a list of namedtuples from the result of a join query . A
 single database result is partitioned at intervals corresponding to the
 fields in namedTuples . The return value is the result of applying
 namedtuple . _ make ( ) to each of the partitions , for each of the namedTuples .
 Parameters :
 result : Tuple representing a single result from a database query
 * namedTuples : List of named tuples ."""

implementation:
results = ClientJobsDAO . partitionAtIntervals ( result , [ len ( nt . _fields ) for nt in namedTuples ] )
return [ nt . _make ( result ) for nt , result in zip ( namedTuples , results ) ]


signature:
def jobInfoWithModels ( self , jobID ) :
    """Get all info about a job , with model details , if available .
 Parameters :
 job : jobID of the job to query
 retval : A sequence of two - tuples if the jobID exists in the jobs
 table ( exeption is raised if it doesn ' t exist ) . Each two - tuple
 contains an instance of jobInfoNamedTuple as the first element and
 an instance of modelInfoNamedTuple as the second element . NOTE : In
 the case where there are no matching model rows , a sequence of one
 two - tuple will still be returned , but the modelInfoNamedTuple
 fields will be None , and the jobInfoNamedTuple fields will be
 populated ."""

implementation:
# Get a database connection and cursor
combinedResults = None
with ConnectionFactory . get ( ) as conn : # NOTE : Since we ' re using a LEFT JOIN on the models table , there need not
# be a matching row in the models table , but the matching row from the
# jobs table will still be returned ( along with all fields from the models
# table with values of None in case there were no matchings models )
    query = ' ' . join ( [ 'SELECT %s.*, %s.*' % ( self . jobsTableName , self . modelsTableName ) , 'FROM %s' % self . jobsTableName , 'LEFT JOIN %s USING(job_id)' % self . modelsTableName , 'WHERE job_id=%s' ] )
    conn . cursor . execute ( query , ( jobID , ) )
    if conn . cursor . rowcount > 0 :
        combinedResults = [ ClientJobsDAO . _combineResults ( result , self . _jobs . jobInfoNamedTuple , self . _models . modelInfoNamedTuple ) for result in conn . cursor . fetchall ( ) ]
if combinedResults is not None :
    return combinedResults
raise RuntimeError ( "jobID=%s not found within the jobs table" % ( jobID ) )


signature:
def jobInfo ( self , jobID ) :
    """Get all info about a job
 Parameters :
 job : jobID of the job to query
 retval : namedtuple containing the job info ."""

implementation:
row = self . _getOneMatchingRowWithRetries ( self . _jobs , dict ( job_id = jobID ) , [ self . _jobs . pubToDBNameDict [ n ] for n in self . _jobs . jobInfoNamedTuple . _fields ] )
if row is None :
    raise RuntimeError ( "jobID=%s not found within the jobs table" % ( jobID ) )
# Create a namedtuple with the names to values
return self . _jobs . jobInfoNamedTuple . _make ( row )


signature:
def jobSetStatus ( self , jobID , status , useConnectionID = True , ) :
    """Change the status on the given job
 Parameters :
 job : jobID of the job to change status
 status : new status string ( ClientJobsDAO . STATUS _ xxxxx )
 useConnectionID : True if the connection id of the calling function
 must be the same as the connection that created the job . Set
 to False for hypersearch workers"""

implementation:
# Get a database connection and cursor
with ConnectionFactory . get ( ) as conn :
    query = 'UPDATE %s SET status=%%s, ' '              _eng_last_update_time=UTC_TIMESTAMP() ' '          WHERE job_id=%%s' % ( self . jobsTableName , )
    sqlParams = [ status , jobID ]
    if useConnectionID :
        query += ' AND _eng_cjm_conn_id=%s'
        sqlParams . append ( self . _connectionID )
    result = conn . cursor . execute ( query , sqlParams )
    if result != 1 :
        raise RuntimeError ( "Tried to change the status of job %d to %s, but " "this job belongs to some other CJM" % ( jobID , status ) )


signature:
def jobSetCompleted ( self , jobID , completionReason , completionMsg , useConnectionID = True ) :
    """Change the status on the given job to completed
 Parameters :
 job : jobID of the job to mark as completed
 completionReason : completionReason string
 completionMsg : completionMsg string
 useConnectionID : True if the connection id of the calling function
 must be the same as the connection that created the job . Set
 to False for hypersearch workers"""

implementation:
# Get a database connection and cursor
with ConnectionFactory . get ( ) as conn :
    query = 'UPDATE %s SET status=%%s, ' '              completion_reason=%%s, ' '              completion_msg=%%s, ' '              end_time=UTC_TIMESTAMP(), ' '              _eng_last_update_time=UTC_TIMESTAMP() ' '          WHERE job_id=%%s' % ( self . jobsTableName , )
    sqlParams = [ self . STATUS_COMPLETED , completionReason , completionMsg , jobID ]
    if useConnectionID :
        query += ' AND _eng_cjm_conn_id=%s'
        sqlParams . append ( self . _connectionID )
    result = conn . cursor . execute ( query , sqlParams )
    if result != 1 :
        raise RuntimeError ( "Tried to change the status of jobID=%s to " "completed, but this job could not be found or " "belongs to some other CJM" % ( jobID ) )


signature:
def jobCancel ( self , jobID ) :
    """Cancel the given job . This will update the cancel field in the
 jobs table and will result in the job being cancelled .
 Parameters :
 jobID : jobID of the job to mark as completed
 to False for hypersearch workers"""

implementation:
self . _logger . info ( 'Canceling jobID=%s' , jobID )
# NOTE : jobSetFields does retries on transient mysql failures
self . jobSetFields ( jobID , { "cancel" : True } , useConnectionID = False )


signature:
def jobGetModelIDs ( self , jobID ) :
    """Fetch all the modelIDs that correspond to a given jobID ; empty sequence
 if none"""

implementation:
rows = self . _getMatchingRowsWithRetries ( self . _models , dict ( job_id = jobID ) , [ 'model_id' ] )
return [ r [ 0 ] for r in rows ]


signature:
def getActiveJobCountForClientInfo ( self , clientInfo ) :
    """Return the number of jobs for the given clientInfo and a status that is
 not completed ."""

implementation:
with ConnectionFactory . get ( ) as conn :
    query = 'SELECT count(job_id) ' 'FROM %s ' 'WHERE client_info = %%s ' ' AND status != %%s' % self . jobsTableName
    conn . cursor . execute ( query , [ clientInfo , self . STATUS_COMPLETED ] )
    activeJobCount = conn . cursor . fetchone ( ) [ 0 ]
return activeJobCount


signature:
def getActiveJobCountForClientKey ( self , clientKey ) :
    """Return the number of jobs for the given clientKey and a status that is
 not completed ."""

implementation:
with ConnectionFactory . get ( ) as conn :
    query = 'SELECT count(job_id) ' 'FROM %s ' 'WHERE client_key = %%s ' ' AND status != %%s' % self . jobsTableName
    conn . cursor . execute ( query , [ clientKey , self . STATUS_COMPLETED ] )
    activeJobCount = conn . cursor . fetchone ( ) [ 0 ]
return activeJobCount


signature:
def getActiveJobsForClientInfo ( self , clientInfo , fields = [ ] ) :
    """Fetch jobIDs for jobs in the table with optional fields given a
 specific clientInfo"""

implementation:
# Form the sequence of field name strings that will go into the
# request
dbFields = [ self . _jobs . pubToDBNameDict [ x ] for x in fields ]
dbFieldsStr = ',' . join ( [ 'job_id' ] + dbFields )
with ConnectionFactory . get ( ) as conn :
    query = 'SELECT %s FROM %s ' 'WHERE client_info = %%s ' ' AND status != %%s' % ( dbFieldsStr , self . jobsTableName )
    conn . cursor . execute ( query , [ clientInfo , self . STATUS_COMPLETED ] )
    rows = conn . cursor . fetchall ( )
return rows


signature:
def getFieldsForActiveJobsOfType ( self , jobType , fields = [ ] ) :
    """Helper function for querying the models table including relevant job
 info where the job type matches the specified jobType . Only records for
 which there is a matching jobId in both tables is returned , and only the
 requested fields are returned in each result , assuming that there is not
 a conflict . This function is useful , for example , in querying a cluster
 for a list of actively running production models ( according to the state
 of the client jobs database ) . jobType must be one of the JOB _ TYPE _ XXXX
 enumerations .
 Parameters :
 jobType : jobType enum
 fields : list of fields to return
 Returns : List of tuples containing the jobId and requested field values"""

implementation:
dbFields = [ self . _jobs . pubToDBNameDict [ x ] for x in fields ]
dbFieldsStr = ',' . join ( [ 'job_id' ] + dbFields )
with ConnectionFactory . get ( ) as conn :
    query = 'SELECT DISTINCT %s ' 'FROM %s j ' 'LEFT JOIN %s m USING(job_id) ' 'WHERE j.status != %%s ' 'AND _eng_job_type = %%s' % ( dbFieldsStr , self . jobsTableName , self . modelsTableName )
    conn . cursor . execute ( query , [ self . STATUS_COMPLETED , jobType ] )
    return conn . cursor . fetchall ( )


signature:
def jobGetFields ( self , jobID , fields ) :
    """Fetch the values of 1 or more fields from a job record . Here , ' fields '
 is a list with the names of the fields to fetch . The names are the public
 names of the fields ( camelBack , not the lower _ case _ only form as stored in
 the DB ) .
 Parameters :
 jobID : jobID of the job record
 fields : list of fields to return
 Returns : A sequence of field values in the same order as the requested
 field list - > [ field1 , field2 , . . . ]"""

implementation:
# NOTE : jobsGetFields retries on transient mysql failures
return self . jobsGetFields ( [ jobID ] , fields , requireAll = True ) [ 0 ] [ 1 ]


signature:
def jobsGetFields ( self , jobIDs , fields , requireAll = True ) :
    """Fetch the values of 1 or more fields from a sequence of job records .
 Here , ' fields ' is a sequence ( list or tuple ) with the names of the fields to
 fetch . The names are the public names of the fields ( camelBack , not the
 lower _ case _ only form as stored in the DB ) .
 WARNING ! ! ! : The order of the results are NOT necessarily in the same order as
 the order of the job IDs passed in ! ! !
 Parameters :
 jobIDs : A sequence of jobIDs
 fields : A list of fields to return for each jobID
 Returns : A list of tuples - > ( jobID , [ field1 , field2 , . . . ] )"""

implementation:
assert isinstance ( jobIDs , self . _SEQUENCE_TYPES )
assert len ( jobIDs ) >= 1
rows = self . _getMatchingRowsWithRetries ( self . _jobs , dict ( job_id = jobIDs ) , [ 'job_id' ] + [ self . _jobs . pubToDBNameDict [ x ] for x in fields ] )
if requireAll and len ( rows ) < len ( jobIDs ) : # NOTE : this will also trigger if the jobIDs list included duplicates
    raise RuntimeError ( "jobIDs %s not found within the jobs table" % ( ( set ( jobIDs ) - set ( r [ 0 ] for r in rows ) ) , ) )
return [ ( r [ 0 ] , list ( r [ 1 : ] ) ) for r in rows ]


signature:
def jobSetFields ( self , jobID , fields , useConnectionID = True , ignoreUnchanged = False ) :
    """Change the values of 1 or more fields in a job . Here , ' fields ' is a
 dict with the name / value pairs to change . The names are the public names of
 the fields ( camelBack , not the lower _ case _ only form as stored in the DB ) .
 This method is for private use by the ClientJobManager only .
 Parameters :
 jobID : jobID of the job record
 fields : dictionary of fields to change
 useConnectionID : True if the connection id of the calling function
 must be the same as the connection that created the job . Set
 to False for hypersearch workers
 ignoreUnchanged : The default behavior is to throw a
 RuntimeError if no rows are affected . This could either be
 because :
 1 ) Because there was not matching jobID
 2 ) or if the data to update matched the data in the DB exactly .
 Set this parameter to True if you expect case 2 and wish to
 supress the error ."""

implementation:
# Form the sequecce of key = value strings that will go into the
# request
assignmentExpressions = ',' . join ( [ "%s=%%s" % ( self . _jobs . pubToDBNameDict [ f ] , ) for f in fields . iterkeys ( ) ] )
assignmentValues = fields . values ( )
query = 'UPDATE %s SET %s ' '          WHERE job_id=%%s' % ( self . jobsTableName , assignmentExpressions , )
sqlParams = assignmentValues + [ jobID ]
if useConnectionID :
    query += ' AND _eng_cjm_conn_id=%s'
    sqlParams . append ( self . _connectionID )
# Get a database connection and cursor
with ConnectionFactory . get ( ) as conn :
    result = conn . cursor . execute ( query , sqlParams )
if result != 1 and not ignoreUnchanged :
    raise RuntimeError ( "Tried to change fields (%r) of jobID=%s conn_id=%r), but an error " "occurred. result=%r; query=%r" % ( assignmentExpressions , jobID , self . _connectionID , result , query ) )


signature:
def jobSetFieldIfEqual ( self , jobID , fieldName , newValue , curValue ) :
    """Change the value of 1 field in a job to ' newValue ' , but only if the
 current value matches ' curValue ' . The ' fieldName ' is the public name of
 the field ( camelBack , not the lower _ case _ only form as stored in the DB ) .
 This method is used for example by HypersearcWorkers to update the
 engWorkerState field periodically . By qualifying on curValue , it insures
 that only 1 worker at a time is elected to perform the next scheduled
 periodic sweep of the models .
 Parameters :
 jobID : jobID of the job record to modify
 fieldName : public field name of the field
 newValue : new value of the field to set
 curValue : current value to qualify against
 retval : True if we successfully modified the field
 False if curValue did not match"""

implementation:
# Get the private field name and string form of the value
dbFieldName = self . _jobs . pubToDBNameDict [ fieldName ]
conditionValue = [ ]
if isinstance ( curValue , bool ) :
    conditionExpression = '%s IS %s' % ( dbFieldName , { True : 'TRUE' , False : 'FALSE' } [ curValue ] )
elif curValue is None :
    conditionExpression = '%s is NULL' % ( dbFieldName , )
else :
    conditionExpression = '%s=%%s' % ( dbFieldName , )
    conditionValue . append ( curValue )
query = 'UPDATE %s SET _eng_last_update_time=UTC_TIMESTAMP(), %s=%%s ' '          WHERE job_id=%%s AND %s' % ( self . jobsTableName , dbFieldName , conditionExpression )
sqlParams = [ newValue , jobID ] + conditionValue
with ConnectionFactory . get ( ) as conn :
    result = conn . cursor . execute ( query , sqlParams )
return ( result == 1 )


signature:
def jobIncrementIntField ( self , jobID , fieldName , increment = 1 , useConnectionID = False ) :
    """Incremet the value of 1 field in a job by increment . The ' fieldName ' is
 the public name of the field ( camelBack , not the lower _ case _ only form as
 stored in the DB ) .
 This method is used for example by HypersearcWorkers to update the
 engWorkerState field periodically . By qualifying on curValue , it insures
 that only 1 worker at a time is elected to perform the next scheduled
 periodic sweep of the models .
 Parameters :
 jobID : jobID of the job record to modify
 fieldName : public field name of the field
 increment : increment is added to the current value of the field"""

implementation:
# Get the private field name and string form of the value
dbFieldName = self . _jobs . pubToDBNameDict [ fieldName ]
# Get a database connection and cursor
with ConnectionFactory . get ( ) as conn :
    query = 'UPDATE %s SET %s=%s+%%s ' '          WHERE job_id=%%s' % ( self . jobsTableName , dbFieldName , dbFieldName )
    sqlParams = [ increment , jobID ]
    if useConnectionID :
        query += ' AND _eng_cjm_conn_id=%s'
        sqlParams . append ( self . _connectionID )
    result = conn . cursor . execute ( query , sqlParams )
if result != 1 :
    raise RuntimeError ( "Tried to increment the field (%r) of jobID=%s (conn_id=%r), but an " "error occurred. result=%r; query=%r" % ( dbFieldName , jobID , self . _connectionID , result , query ) )


signature:
def jobUpdateResults ( self , jobID , results ) :
    """Update the results string and last - update - time fields of a model .
 Parameters :
 jobID : job ID of model to modify
 results : new results ( json dict string )"""

implementation:
with ConnectionFactory . get ( ) as conn :
    query = 'UPDATE %s SET _eng_last_update_time=UTC_TIMESTAMP(), ' '              results=%%s ' '          WHERE job_id=%%s' % ( self . jobsTableName , )
    conn . cursor . execute ( query , [ results , jobID ] )


signature:
def modelsClearAll ( self ) :
    """Delete all models from the models table
 Parameters :"""

implementation:
self . _logger . info ( 'Deleting all rows from models table %r' , self . modelsTableName )
with ConnectionFactory . get ( ) as conn :
    query = 'DELETE FROM %s' % ( self . modelsTableName )
    conn . cursor . execute ( query )


signature:
def modelsInfo ( self , modelIDs ) :
    """Get ALL info for a set of models
 WARNING ! ! ! : The order of the results are NOT necessarily in the same order as
 the order of the model IDs passed in ! ! !
 Parameters :
 modelIDs : list of model IDs
 retval : list of nametuples containing all the fields stored for each
 model ."""

implementation:
assert isinstance ( modelIDs , self . _SEQUENCE_TYPES ) , ( "wrong modelIDs type: %s" ) % ( type ( modelIDs ) , )
assert modelIDs , "modelIDs is empty"
rows = self . _getMatchingRowsWithRetries ( self . _models , dict ( model_id = modelIDs ) , [ self . _models . pubToDBNameDict [ f ] for f in self . _models . modelInfoNamedTuple . _fields ] )
results = [ self . _models . modelInfoNamedTuple . _make ( r ) for r in rows ]
# NOTE : assetion will also fail if modelIDs contains duplicates
assert len ( results ) == len ( modelIDs ) , "modelIDs not found: %s" % ( set ( modelIDs ) - set ( r . modelId for r in results ) )
return results


signature:
def modelsGetFields ( self , modelIDs , fields ) :
    """Fetch the values of 1 or more fields from a sequence of model records .
 Here , ' fields ' is a list with the names of the fields to fetch . The names
 are the public names of the fields ( camelBack , not the lower _ case _ only form
 as stored in the DB ) .
 WARNING ! ! ! : The order of the results are NOT necessarily in the same order
 as the order of the model IDs passed in ! ! !
 Parameters :
 modelIDs : A single modelID or sequence of modelIDs
 fields : A list of fields to return
 Returns : If modelIDs is a sequence :
 a list of tuples - > ( modelID , [ field1 , field2 , . . . ] )
 If modelIDs is a single modelID :
 a list of field values - > [ field1 , field2 , . . . ]"""

implementation:
assert len ( fields ) >= 1 , 'fields is empty'
# Form the sequence of field name strings that will go into the
# request
isSequence = isinstance ( modelIDs , self . _SEQUENCE_TYPES )
if isSequence :
    assert len ( modelIDs ) >= 1 , 'modelIDs is empty'
else :
    modelIDs = [ modelIDs ]
rows = self . _getMatchingRowsWithRetries ( self . _models , dict ( model_id = modelIDs ) , [ 'model_id' ] + [ self . _models . pubToDBNameDict [ f ] for f in fields ] )
if len ( rows ) < len ( modelIDs ) :
    raise RuntimeError ( "modelIDs not found within the models table: %s" % ( ( set ( modelIDs ) - set ( r [ 0 ] for r in rows ) ) , ) )
if not isSequence :
    return list ( rows [ 0 ] [ 1 : ] )
return [ ( r [ 0 ] , list ( r [ 1 : ] ) ) for r in rows ]


signature:
def modelsGetFieldsForJob ( self , jobID , fields , ignoreKilled = False ) :
    """Gets the specified fields for all the models for a single job . This is
 similar to modelsGetFields
 Parameters :
 jobID : jobID for the models to be searched
 fields : A list of fields to return
 ignoreKilled : ( True / False ) . If True , this will ignore models that
 have been killed
 Returns : a ( possibly empty ) list of tuples as follows
 ( model _ id1 , [ field1 , . . . , fieldn ] ) ,
 ( model _ id2 , [ field1 , . . . , fieldn ] ) ,
 ( model _ id3 , [ field1 , . . . , fieldn ] )
 NOTE : since there is a window of time between a job getting inserted into
 jobs table and the job ' s worker ( s ) starting up and creating models , an
 empty - list result is one of the normal outcomes ."""

implementation:
assert len ( fields ) >= 1 , 'fields is empty'
# Form the sequence of field name strings that will go into the
# request
dbFields = [ self . _models . pubToDBNameDict [ x ] for x in fields ]
dbFieldsStr = ',' . join ( dbFields )
query = 'SELECT model_id, %s FROM %s ' '          WHERE job_id=%%s ' % ( dbFieldsStr , self . modelsTableName )
sqlParams = [ jobID ]
if ignoreKilled :
    query += ' AND (completion_reason IS NULL OR completion_reason != %s)'
    sqlParams . append ( self . CMPL_REASON_KILLED )
# Get a database connection and cursor
with ConnectionFactory . get ( ) as conn :
    conn . cursor . execute ( query , sqlParams )
    rows = conn . cursor . fetchall ( )
if rows is None : # fetchall is defined to return a ( possibly - empty ) sequence of
# sequences ; however , we occasionally see None returned and don ' t know
# why . . .
    self . _logger . error ( "Unexpected None result from cursor.fetchall; " "query=%r; Traceback=%r" , query , traceback . format_exc ( ) )
return [ ( r [ 0 ] , list ( r [ 1 : ] ) ) for r in rows ]


signature:
def modelsGetFieldsForCheckpointed ( self , jobID , fields ) :
    """Gets fields from all models in a job that have been checkpointed . This is
 used to figure out whether or not a new model should be checkpointed .
 Parameters :
 jobID : The jobID for the models to be searched
 fields : A list of fields to return
 Returns : a ( possibly - empty ) list of tuples as follows
 ( model _ id1 , [ field1 , . . . , fieldn ] ) ,
 ( model _ id2 , [ field1 , . . . , fieldn ] ) ,
 ( model _ id3 , [ field1 , . . . , fieldn ] )"""

implementation:
assert len ( fields ) >= 1 , "fields is empty"
# Get a database connection and cursor
with ConnectionFactory . get ( ) as conn :
    dbFields = [ self . _models . pubToDBNameDict [ f ] for f in fields ]
    dbFieldStr = ", " . join ( dbFields )
    query = 'SELECT model_id, {fields} from {models}' '   WHERE job_id=%s AND model_checkpoint_id IS NOT NULL' . format ( fields = dbFieldStr , models = self . modelsTableName )
    conn . cursor . execute ( query , [ jobID ] )
    rows = conn . cursor . fetchall ( )
return [ ( r [ 0 ] , list ( r [ 1 : ] ) ) for r in rows ]


signature:
def modelSetFields ( self , modelID , fields , ignoreUnchanged = False ) :
    """Change the values of 1 or more fields in a model . Here , ' fields ' is a
 dict with the name / value pairs to change . The names are the public names of
 the fields ( camelBack , not the lower _ case _ only form as stored in the DB ) .
 Parameters :
 jobID : jobID of the job record
 fields : dictionary of fields to change
 ignoreUnchanged : The default behavior is to throw a
 RuntimeError if no rows are affected . This could either be
 because :
 1 ) Because there was no matching modelID
 2 ) or if the data to update matched the data in the DB exactly .
 Set this parameter to True if you expect case 2 and wish to
 supress the error ."""

implementation:
# Form the sequence of key = value strings that will go into the
# request
assignmentExpressions = ',' . join ( '%s=%%s' % ( self . _models . pubToDBNameDict [ f ] , ) for f in fields . iterkeys ( ) )
assignmentValues = fields . values ( )
query = 'UPDATE %s SET %s, update_counter = update_counter+1 ' '          WHERE model_id=%%s' % ( self . modelsTableName , assignmentExpressions )
sqlParams = assignmentValues + [ modelID ]
# Get a database connection and cursor
with ConnectionFactory . get ( ) as conn :
    numAffectedRows = conn . cursor . execute ( query , sqlParams )
    self . _logger . debug ( "Executed: numAffectedRows=%r, query=%r, sqlParams=%r" , numAffectedRows , query , sqlParams )
if numAffectedRows != 1 and not ignoreUnchanged :
    raise RuntimeError ( ( "Tried to change fields (%r) of model %r (conn_id=%r), but an error " "occurred. numAffectedRows=%r; query=%r; sqlParams=%r" ) % ( fields , modelID , self . _connectionID , numAffectedRows , query , sqlParams , ) )


signature:
def modelsGetParams ( self , modelIDs ) :
    """Get the params and paramsHash for a set of models .
 WARNING ! ! ! : The order of the results are NOT necessarily in the same order as
 the order of the model IDs passed in ! ! !
 Parameters :
 modelIDs : list of model IDs
 retval : list of result namedtuples defined in
 ClientJobsDAO . _ models . getParamsNamedTuple . Each tuple
 contains : ( modelId , params , engParamsHash )"""

implementation:
assert isinstance ( modelIDs , self . _SEQUENCE_TYPES ) , ( "Wrong modelIDs type: %r" ) % ( type ( modelIDs ) , )
assert len ( modelIDs ) >= 1 , "modelIDs is empty"
rows = self . _getMatchingRowsWithRetries ( self . _models , { 'model_id' : modelIDs } , [ self . _models . pubToDBNameDict [ f ] for f in self . _models . getParamsNamedTuple . _fields ] )
# NOTE : assertion will also fail when modelIDs contains duplicates
assert len ( rows ) == len ( modelIDs ) , "Didn't find modelIDs: %r" % ( ( set ( modelIDs ) - set ( r [ 0 ] for r in rows ) ) , )
# Return the params and params hashes as a namedtuple
return [ self . _models . getParamsNamedTuple . _make ( r ) for r in rows ]


signature:
def modelsGetResultAndStatus ( self , modelIDs ) :
    """Get the results string and other status fields for a set of models .
 WARNING ! ! ! : The order of the results are NOT necessarily in the same order
 as the order of the model IDs passed in ! ! !
 For each model , this returns a tuple containing :
 ( modelID , results , status , updateCounter , numRecords , completionReason ,
 completionMsg , engParamsHash
 Parameters :
 modelIDs : list of model IDs
 retval : list of result tuples . Each tuple contains :
 ( modelID , results , status , updateCounter , numRecords ,
 completionReason , completionMsg , engParamsHash )"""

implementation:
assert isinstance ( modelIDs , self . _SEQUENCE_TYPES ) , ( "Wrong modelIDs type: %r" ) % type ( modelIDs )
assert len ( modelIDs ) >= 1 , "modelIDs is empty"
rows = self . _getMatchingRowsWithRetries ( self . _models , { 'model_id' : modelIDs } , [ self . _models . pubToDBNameDict [ f ] for f in self . _models . getResultAndStatusNamedTuple . _fields ] )
# NOTE : assertion will also fail when modelIDs contains duplicates
assert len ( rows ) == len ( modelIDs ) , "Didn't find modelIDs: %r" % ( ( set ( modelIDs ) - set ( r [ 0 ] for r in rows ) ) , )
# Return the results as a list of namedtuples
return [ self . _models . getResultAndStatusNamedTuple . _make ( r ) for r in rows ]


signature:
def modelsGetUpdateCounters ( self , jobID ) :
    """Return info on all of the models that are in already in the models
 table for a given job . For each model , this returns a tuple
 containing : ( modelID , updateCounter ) .
 Note that we don ' t return the results for all models , since the results
 string could be quite large . The information we are returning is
 just 2 integer fields .
 Parameters :
 jobID : jobID to query
 retval : ( possibly empty ) list of tuples . Each tuple contains :
 ( modelID , updateCounter )"""

implementation:
rows = self . _getMatchingRowsWithRetries ( self . _models , { 'job_id' : jobID } , [ self . _models . pubToDBNameDict [ f ] for f in self . _models . getUpdateCountersNamedTuple . _fields ] )
# Return the results as a list of namedtuples
return [ self . _models . getUpdateCountersNamedTuple . _make ( r ) for r in rows ]


signature:
def modelUpdateResults ( self , modelID , results = None , metricValue = None , numRecords = None ) :
    """Update the results string , and / or num _ records fields of
 a model . This will fail if the model does not currently belong to this
 client ( connection _ id doesn ' t match ) .
 Parameters :
 modelID : model ID of model to modify
 results : new results , or None to ignore
 metricValue : the value of the metric being optimized , or None to ignore
 numRecords : new numRecords , or None to ignore"""

implementation:
assignmentExpressions = [ '_eng_last_update_time=UTC_TIMESTAMP()' , 'update_counter=update_counter+1' ]
assignmentValues = [ ]
if results is not None :
    assignmentExpressions . append ( 'results=%s' )
    assignmentValues . append ( results )
if numRecords is not None :
    assignmentExpressions . append ( 'num_records=%s' )
    assignmentValues . append ( numRecords )
# NOTE1 : ( metricValue = = metricValue ) tests for Nan
# NOTE2 : metricValue is being passed as numpy . float64
if metricValue is not None and ( metricValue == metricValue ) :
    assignmentExpressions . append ( 'optimized_metric=%s' )
    assignmentValues . append ( float ( metricValue ) )
query = 'UPDATE %s SET %s ' '          WHERE model_id=%%s and _eng_worker_conn_id=%%s' % ( self . modelsTableName , ',' . join ( assignmentExpressions ) )
sqlParams = assignmentValues + [ modelID , self . _connectionID ]
# Get a database connection and cursor
with ConnectionFactory . get ( ) as conn :
    numRowsAffected = conn . cursor . execute ( query , sqlParams )
if numRowsAffected != 1 :
    raise InvalidConnectionException ( ( "Tried to update the info of modelID=%r using connectionID=%r, but " "this model belongs to some other worker or modelID not found; " "numRowsAffected=%r" ) % ( modelID , self . _connectionID , numRowsAffected , ) )