signature:
def write ( self , proto ) :
    """Writes serialized data to proto object .
 : param proto : ( DynamicStructBuilder ) Proto object"""

implementation:
# capnp fails to save a tuple . Let ' s force columnDimensions to list .
proto . columnDimensions = list ( self . columnDimensions )
proto . cellsPerColumn = self . cellsPerColumn
proto . activationThreshold = self . activationThreshold
proto . initialPermanence = round ( self . initialPermanence , EPSILON_ROUND )
proto . connectedPermanence = round ( self . connectedPermanence , EPSILON_ROUND )
proto . minThreshold = self . minThreshold
proto . maxNewSynapseCount = self . maxNewSynapseCount
proto . permanenceIncrement = round ( self . permanenceIncrement , EPSILON_ROUND )
proto . permanenceDecrement = round ( self . permanenceDecrement , EPSILON_ROUND )
proto . predictedSegmentDecrement = self . predictedSegmentDecrement
proto . maxSegmentsPerCell = self . maxSegmentsPerCell
proto . maxSynapsesPerSegment = self . maxSynapsesPerSegment
self . connections . write ( proto . connections )
self . _random . write ( proto . random )
proto . activeCells = list ( self . activeCells )
proto . winnerCells = list ( self . winnerCells )
protoActiveSegments = proto . init ( "activeSegments" , len ( self . activeSegments ) )
for i , segment in enumerate ( self . activeSegments ) :
    protoActiveSegments [ i ] . cell = segment . cell
    idx = self . connections . segmentsForCell ( segment . cell ) . index ( segment )
    protoActiveSegments [ i ] . idxOnCell = idx
protoMatchingSegments = proto . init ( "matchingSegments" , len ( self . matchingSegments ) )
for i , segment in enumerate ( self . matchingSegments ) :
    protoMatchingSegments [ i ] . cell = segment . cell
    idx = self . connections . segmentsForCell ( segment . cell ) . index ( segment )
    protoMatchingSegments [ i ] . idxOnCell = idx
protoNumActivePotential = proto . init ( "numActivePotentialSynapsesForSegment" , len ( self . numActivePotentialSynapsesForSegment ) )
for i , numActivePotentialSynapses in enumerate ( self . numActivePotentialSynapsesForSegment ) :
    segment = self . connections . segmentForFlatIdx ( i )
    if segment is not None :
        protoNumActivePotential [ i ] . cell = segment . cell
        idx = self . connections . segmentsForCell ( segment . cell ) . index ( segment )
        protoNumActivePotential [ i ] . idxOnCell = idx
        protoNumActivePotential [ i ] . number = numActivePotentialSynapses
proto . iteration = self . iteration
protoLastUsedIteration = proto . init ( "lastUsedIterationForSegment" , len ( self . numActivePotentialSynapsesForSegment ) )
for i , lastUsed in enumerate ( self . lastUsedIterationForSegment ) :
    segment = self . connections . segmentForFlatIdx ( i )
    if segment is not None :
        protoLastUsedIteration [ i ] . cell = segment . cell
        idx = self . connections . segmentsForCell ( segment . cell ) . index ( segment )
        protoLastUsedIteration [ i ] . idxOnCell = idx
        protoLastUsedIteration [ i ] . number = lastUsed


signature:
def read ( cls , proto ) :
    """Reads deserialized data from proto object .
 : param proto : ( DynamicStructBuilder ) Proto object
 : returns : ( : class : TemporalMemory ) TemporalMemory instance"""

implementation:
tm = object . __new__ ( cls )
# capnp fails to save a tuple , so proto . columnDimensions was forced to
# serialize as a list . We prefer a tuple , however , because columnDimensions
# should be regarded as immutable .
tm . columnDimensions = tuple ( proto . columnDimensions )
tm . cellsPerColumn = int ( proto . cellsPerColumn )
tm . activationThreshold = int ( proto . activationThreshold )
tm . initialPermanence = round ( proto . initialPermanence , EPSILON_ROUND )
tm . connectedPermanence = round ( proto . connectedPermanence , EPSILON_ROUND )
tm . minThreshold = int ( proto . minThreshold )
tm . maxNewSynapseCount = int ( proto . maxNewSynapseCount )
tm . permanenceIncrement = round ( proto . permanenceIncrement , EPSILON_ROUND )
tm . permanenceDecrement = round ( proto . permanenceDecrement , EPSILON_ROUND )
tm . predictedSegmentDecrement = round ( proto . predictedSegmentDecrement , EPSILON_ROUND )
tm . maxSegmentsPerCell = int ( proto . maxSegmentsPerCell )
tm . maxSynapsesPerSegment = int ( proto . maxSynapsesPerSegment )
tm . connections = Connections . read ( proto . connections )
# pylint : disable = W0212
tm . _random = Random ( )
tm . _random . read ( proto . random )
# pylint : enable = W0212
tm . activeCells = [ int ( x ) for x in proto . activeCells ]
tm . winnerCells = [ int ( x ) for x in proto . winnerCells ]
flatListLength = tm . connections . segmentFlatListLength ( )
tm . numActiveConnectedSynapsesForSegment = [ 0 ] * flatListLength
tm . numActivePotentialSynapsesForSegment = [ 0 ] * flatListLength
tm . lastUsedIterationForSegment = [ 0 ] * flatListLength
tm . activeSegments = [ ]
tm . matchingSegments = [ ]
for protoSegment in proto . activeSegments :
    tm . activeSegments . append ( tm . connections . getSegment ( protoSegment . cell , protoSegment . idxOnCell ) )
for protoSegment in proto . matchingSegments :
    tm . matchingSegments . append ( tm . connections . getSegment ( protoSegment . cell , protoSegment . idxOnCell ) )
for protoSegment in proto . numActivePotentialSynapsesForSegment :
    segment = tm . connections . getSegment ( protoSegment . cell , protoSegment . idxOnCell )
    tm . numActivePotentialSynapsesForSegment [ segment . flatIdx ] = ( int ( protoSegment . number ) )
tm . iteration = long ( proto . iteration )
for protoSegment in proto . lastUsedIterationForSegment :
    segment = tm . connections . getSegment ( protoSegment . cell , protoSegment . idxOnCell )
    tm . lastUsedIterationForSegment [ segment . flatIdx ] = ( long ( protoSegment . number ) )
return tm


signature:
def generateFromNumbers ( self , numbers ) :
    """Generate a sequence from a list of numbers .
 Note : Any ` None ` in the list of numbers is considered a reset .
 @ param numbers ( list ) List of numbers
 @ return ( list ) Generated sequence"""

implementation:
sequence = [ ]
for number in numbers :
    if number == None :
        sequence . append ( number )
    else :
        pattern = self . patternMachine . get ( number )
        sequence . append ( pattern )
return sequence


signature:
def addSpatialNoise ( self , sequence , amount ) :
    """Add spatial noise to each pattern in the sequence .
 @ param sequence ( list ) Sequence
 @ param amount ( float ) Amount of spatial noise
 @ return ( list ) Sequence with spatial noise"""

implementation:
newSequence = [ ]
for pattern in sequence :
    if pattern is not None :
        pattern = self . patternMachine . addNoise ( pattern , amount )
    newSequence . append ( pattern )
return newSequence


signature:
def prettyPrintSequence ( self , sequence , verbosity = 1 ) :
    """Pretty print a sequence .
 @ param sequence ( list ) Sequence
 @ param verbosity ( int ) Verbosity level
 @ return ( string ) Pretty - printed text"""

implementation:
text = ""
for i in xrange ( len ( sequence ) ) :
    pattern = sequence [ i ]
    if pattern == None :
        text += "<reset>"
        if i < len ( sequence ) - 1 :
            text += "\n"
    else :
        text += self . patternMachine . prettyPrintPattern ( pattern , verbosity = verbosity )
return text


signature:
def ROCCurve ( y_true , y_score ) :
    """compute Receiver operating characteristic ( ROC )
 Note : this implementation is restricted to the binary classification task .
 Parameters
 y _ true : array , shape = [ n _ samples ]
 true binary labels
 y _ score : array , shape = [ n _ samples ]
 target scores , can either be probability estimates of
 the positive class , confidence values , or binary decisions .
 Returns
 fpr : array , shape = [ > 2]
 False Positive Rates
 tpr : array , shape = [ > 2]
 True Positive Rates
 thresholds : array , shape = [ > 2]
 Thresholds on y _ score used to compute fpr and tpr
 Examples
 > > > import numpy as np
 > > > from sklearn import metrics
 > > > y = np . array ( [ 1 , 1 , 2 , 2 ] )
 > > > scores = np . array ( [ 0.1 , 0.4 , 0.35 , 0.8 ] )
 > > > fpr , tpr , thresholds = metrics . roc _ curve ( y , scores )
 > > > fpr
 array ( [ 0 . , 0.5 , 0.5 , 1 . ] )
 References
 http : / / en . wikipedia . org / wiki / Receiver _ operating _ characteristic"""

implementation:
y_true = np . ravel ( y_true )
classes = np . unique ( y_true )
# ROC only for binary classification
if classes . shape [ 0 ] != 2 :
    raise ValueError ( "ROC is defined for binary classification only" )
y_score = np . ravel ( y_score )
n_pos = float ( np . sum ( y_true == classes [ 1 ] ) )
# nb of true positive
n_neg = float ( np . sum ( y_true == classes [ 0 ] ) )
# nb of true negative
thresholds = np . unique ( y_score )
neg_value , pos_value = classes [ 0 ] , classes [ 1 ]
tpr = np . empty ( thresholds . size , dtype = np . float )
# True positive rate
fpr = np . empty ( thresholds . size , dtype = np . float )
# False positive rate
# Build tpr / fpr vector
current_pos_count = current_neg_count = sum_pos = sum_neg = idx = 0
signal = np . c_ [ y_score , y_true ]
sorted_signal = signal [ signal [ : , 0 ] . argsort ( ) , : ] [ : : - 1 ]
last_score = sorted_signal [ 0 ] [ 0 ]
for score , value in sorted_signal :
    if score == last_score :
        if value == pos_value :
            current_pos_count += 1
        else :
            current_neg_count += 1
    else :
        tpr [ idx ] = ( sum_pos + current_pos_count ) / n_pos
        fpr [ idx ] = ( sum_neg + current_neg_count ) / n_neg
        sum_pos += current_pos_count
        sum_neg += current_neg_count
        current_pos_count = 1 if value == pos_value else 0
        current_neg_count = 1 if value == neg_value else 0
        idx += 1
        last_score = score
else :
    tpr [ - 1 ] = ( sum_pos + current_pos_count ) / n_pos
    fpr [ - 1 ] = ( sum_neg + current_neg_count ) / n_neg
# hard decisions , add ( 0,0)
if fpr . shape [ 0 ] == 2 :
    fpr = np . array ( [ 0.0 , fpr [ 0 ] , fpr [ 1 ] ] )
    tpr = np . array ( [ 0.0 , tpr [ 0 ] , tpr [ 1 ] ] )
# trivial decisions , add ( 0,0 ) and ( 1,1)
elif fpr . shape [ 0 ] == 1 :
    fpr = np . array ( [ 0.0 , fpr [ 0 ] , 1.0 ] )
    tpr = np . array ( [ 0.0 , tpr [ 0 ] , 1.0 ] )
return fpr , tpr , thresholds


signature:
def AreaUnderCurve ( x , y ) :
    """Compute Area Under the Curve ( AUC ) using the trapezoidal rule
 Parameters
 x : array , shape = [ n ]
 x coordinates
 y : array , shape = [ n ]
 y coordinates
 Returns
 auc : float
 Examples
 > > > import numpy as np
 > > > from sklearn import metrics
 > > > y = np . array ( [ 1 , 1 , 2 , 2 ] )
 > > > pred = np . array ( [ 0.1 , 0.4 , 0.35 , 0.8 ] )
 > > > fpr , tpr , thresholds = metrics . roc _ curve ( y , pred )
 > > > metrics . auc ( fpr , tpr )
 0.75"""

implementation:
# x , y = check _ arrays ( x , y )
if x . shape [ 0 ] != y . shape [ 0 ] :
    raise ValueError ( 'x and y should have the same shape' ' to compute area under curve,' ' but x.shape = %s and y.shape = %s.' % ( x . shape , y . shape ) )
if x . shape [ 0 ] < 2 :
    raise ValueError ( 'At least 2 points are needed to compute' ' area under curve, but x.shape = %s' % x . shape )
# reorder the data points according to the x axis
order = np . argsort ( x )
x = x [ order ]
y = y [ order ]
h = np . diff ( x )
area = np . sum ( h * ( y [ 1 : ] + y [ : - 1 ] ) ) / 2.0
return area


signature:
def mmPrettyPrintTraces ( traces , breakOnResets = None ) :
    """Returns pretty - printed table of traces .
 @ param traces ( list ) Traces to print in table
 @ param breakOnResets ( BoolsTrace ) Trace of resets to break table on
 @ return ( string ) Pretty - printed table of traces ."""

implementation:
assert len ( traces ) > 0 , "No traces found"
table = PrettyTable ( [ "#" ] + [ trace . prettyPrintTitle ( ) for trace in traces ] )
for i in xrange ( len ( traces [ 0 ] . data ) ) :
    if breakOnResets and breakOnResets . data [ i ] :
        table . add_row ( [ "<reset>" ] * ( len ( traces ) + 1 ) )
    table . add_row ( [ i ] + [ trace . prettyPrintDatum ( trace . data [ i ] ) for trace in traces ] )
return table . get_string ( ) . encode ( "utf-8" )


signature:
def mmPrettyPrintMetrics ( metrics , sigFigs = 5 ) :
    """Returns pretty - printed table of metrics .
 @ param metrics ( list ) Traces to print in table
 @ param sigFigs ( int ) Number of significant figures to print
 @ return ( string ) Pretty - printed table of metrics ."""

implementation:
assert len ( metrics ) > 0 , "No metrics found"
table = PrettyTable ( [ "Metric" , "mean" , "standard deviation" , "min" , "max" , "sum" , ] )
for metric in metrics :
    table . add_row ( [ metric . prettyPrintTitle ( ) ] + metric . getStats ( ) )
return table . get_string ( ) . encode ( "utf-8" )


signature:
def mmGetCellTracePlot ( self , cellTrace , cellCount , activityType , title = "" , showReset = False , resetShading = 0.25 ) :
    """Returns plot of the cell activity . Note that if many timesteps of
 activities are input , matplotlib ' s image interpolation may omit activities
 ( columns in the image ) .
 @ param cellTrace ( list ) a temporally ordered list of sets of cell
 activities
 @ param cellCount ( int ) number of cells in the space being rendered
 @ param activityType ( string ) type of cell activity being displayed
 @ param title ( string ) an optional title for the figure
 @ param showReset ( bool ) if true , the first set of cell activities
 after a reset will have a grayscale background
 @ param resetShading ( float ) applicable if showReset is true , specifies the
 intensity of the reset background with 0.0
 being white and 1.0 being black
 @ return ( Plot ) plot"""

implementation:
plot = Plot ( self , title )
resetTrace = self . mmGetTraceResets ( ) . data
data = numpy . zeros ( ( cellCount , 1 ) )
for i in xrange ( len ( cellTrace ) ) : # Set up a " background " vector that is shaded or blank
    if showReset and resetTrace [ i ] :
        activity = numpy . ones ( ( cellCount , 1 ) ) * resetShading
    else :
        activity = numpy . zeros ( ( cellCount , 1 ) )
    activeIndices = cellTrace [ i ]
    activity [ list ( activeIndices ) ] = 1
    data = numpy . concatenate ( ( data , activity ) , 1 )
plot . add2DArray ( data , xlabel = "Time" , ylabel = activityType , name = title )
return plot


signature:
def estimateAnomalyLikelihoods ( anomalyScores , averagingWindow = 10 , skipRecords = 0 , verbosity = 0 ) :
    """Given a series of anomaly scores , compute the likelihood for each score . This
 function should be called once on a bunch of historical anomaly scores for an
 initial estimate of the distribution . It should be called again every so often
 ( say every 50 records ) to update the estimate .
 : param anomalyScores : a list of records . Each record is a list with the
 following three elements : [ timestamp , value , score ]
 Example : :
 [ datetime . datetime ( 2013 , 8 , 10 , 23 , 0 ) , 6.0 , 1.0]
 For best results , the list should be between 1000
 and 10,000 records
 : param averagingWindow : integer number of records to average over
 : param skipRecords : integer specifying number of records to skip when
 estimating distributions . If skip records are > =
 len ( anomalyScores ) , a very broad distribution is returned
 that makes everything pretty likely .
 : param verbosity : integer controlling extent of printouts for debugging
 0 = none
 1 = occasional information
 2 = print every record
 : returns : 3 - tuple consisting of :
 - likelihoods
 numpy array of likelihoods , one for each aggregated point
 - avgRecordList
 list of averaged input records
 - params
 a small JSON dict that contains the state of the estimator"""

implementation:
if verbosity > 1 :
    print ( "In estimateAnomalyLikelihoods." )
    print ( "Number of anomaly scores:" , len ( anomalyScores ) )
    print ( "Skip records=" , skipRecords )
    print ( "First 20:" , anomalyScores [ 0 : min ( 20 , len ( anomalyScores ) ) ] )
if len ( anomalyScores ) == 0 :
    raise ValueError ( "Must have at least one anomalyScore" )
# Compute averaged anomaly scores
aggRecordList , historicalValues , total = _anomalyScoreMovingAverage ( anomalyScores , windowSize = averagingWindow , verbosity = verbosity )
s = [ r [ 2 ] for r in aggRecordList ]
dataValues = numpy . array ( s )
# Estimate the distribution of anomaly scores based on aggregated records
if len ( aggRecordList ) <= skipRecords :
    distributionParams = nullDistribution ( verbosity = verbosity )
else :
    distributionParams = estimateNormal ( dataValues [ skipRecords : ] )
    # HACK ALERT ! The HTMPredictionModel currently does not handle constant
    # metric values very well ( time of day encoder changes sometimes lead to
    # unstable SDR ' s even though the metric is constant ) . Until this is
    # resolved , we explicitly detect and handle completely flat metric values by
    # reporting them as not anomalous .
    s = [ r [ 1 ] for r in aggRecordList ]
    # Only do this if the values are numeric
    if all ( [ isinstance ( r [ 1 ] , numbers . Number ) for r in aggRecordList ] ) :
        metricValues = numpy . array ( s )
        metricDistribution = estimateNormal ( metricValues [ skipRecords : ] , performLowerBoundCheck = False )
        if metricDistribution [ "variance" ] < 1.5e-5 :
            distributionParams = nullDistribution ( verbosity = verbosity )
# Estimate likelihoods based on this distribution
likelihoods = numpy . array ( dataValues , dtype = float )
for i , s in enumerate ( dataValues ) :
    likelihoods [ i ] = tailProbability ( s , distributionParams )
# Filter likelihood values
filteredLikelihoods = numpy . array ( _filterLikelihoods ( likelihoods ) )
params = { "distribution" : distributionParams , "movingAverage" : { "historicalValues" : historicalValues , "total" : total , "windowSize" : averagingWindow , } , "historicalLikelihoods" : list ( likelihoods [ - min ( averagingWindow , len ( likelihoods ) ) : ] ) , }
if verbosity > 1 :
    print ( "Discovered params=" )
    print ( params )
    print ( "Number of likelihoods:" , len ( likelihoods ) )
    print ( "First 20 likelihoods:" , ( filteredLikelihoods [ 0 : min ( 20 , len ( filteredLikelihoods ) ) ] ) )
    print ( "leaving estimateAnomalyLikelihoods" )
return ( filteredLikelihoods , aggRecordList , params )


signature:
def updateAnomalyLikelihoods ( anomalyScores , params , verbosity = 0 ) :
    """Compute updated probabilities for anomalyScores using the given params .
 : param anomalyScores : a list of records . Each record is a list with the
 following three elements : [ timestamp , value , score ]
 Example : :
 [ datetime . datetime ( 2013 , 8 , 10 , 23 , 0 ) , 6.0 , 1.0]
 : param params : the JSON dict returned by estimateAnomalyLikelihoods
 : param verbosity : integer controlling extent of printouts for debugging
 : type verbosity : int
 : returns : 3 - tuple consisting of :
 - likelihoods
 numpy array of likelihoods , one for each aggregated point
 - avgRecordList
 list of averaged input records
 - params
 an updated JSON object containing the state of this metric ."""

implementation:
if verbosity > 3 :
    print ( "In updateAnomalyLikelihoods." )
    print ( "Number of anomaly scores:" , len ( anomalyScores ) )
    print ( "First 20:" , anomalyScores [ 0 : min ( 20 , len ( anomalyScores ) ) ] )
    print ( "Params:" , params )
if len ( anomalyScores ) == 0 :
    raise ValueError ( "Must have at least one anomalyScore" )
if not isValidEstimatorParams ( params ) :
    raise ValueError ( "'params' is not a valid params structure" )
# For backward compatibility .
if "historicalLikelihoods" not in params :
    params [ "historicalLikelihoods" ] = [ 1.0 ]
# Compute moving averages of these new scores using the previous values
# as well as likelihood for these scores using the old estimator
historicalValues = params [ "movingAverage" ] [ "historicalValues" ]
total = params [ "movingAverage" ] [ "total" ]
windowSize = params [ "movingAverage" ] [ "windowSize" ]
aggRecordList = numpy . zeros ( len ( anomalyScores ) , dtype = float )
likelihoods = numpy . zeros ( len ( anomalyScores ) , dtype = float )
for i , v in enumerate ( anomalyScores ) :
    newAverage , historicalValues , total = ( MovingAverage . compute ( historicalValues , total , v [ 2 ] , windowSize ) )
    aggRecordList [ i ] = newAverage
    likelihoods [ i ] = tailProbability ( newAverage , params [ "distribution" ] )
# Filter the likelihood values . First we prepend the historical likelihoods
# to the current set . Then we filter the values . We peel off the likelihoods
# to return and the last windowSize values to store for later .
likelihoods2 = params [ "historicalLikelihoods" ] + list ( likelihoods )
filteredLikelihoods = _filterLikelihoods ( likelihoods2 )
likelihoods [ : ] = filteredLikelihoods [ - len ( likelihoods ) : ]
historicalLikelihoods = likelihoods2 [ - min ( windowSize , len ( likelihoods2 ) ) : ]
# Update the estimator
newParams = { "distribution" : params [ "distribution" ] , "movingAverage" : { "historicalValues" : historicalValues , "total" : total , "windowSize" : windowSize , } , "historicalLikelihoods" : historicalLikelihoods , }
assert len ( newParams [ "historicalLikelihoods" ] ) <= windowSize
if verbosity > 3 :
    print ( "Number of likelihoods:" , len ( likelihoods ) )
    print ( "First 20 likelihoods:" , likelihoods [ 0 : min ( 20 , len ( likelihoods ) ) ] )
    print ( "Leaving updateAnomalyLikelihoods." )
return ( likelihoods , aggRecordList , newParams )


signature:
def _filterLikelihoods ( likelihoods , redThreshold = 0.99999 , yellowThreshold = 0.999 ) :
    """Filter the list of raw ( pre - filtered ) likelihoods so that we only preserve
 sharp increases in likelihood . ' likelihoods ' can be a numpy array of floats or
 a list of floats .
 : returns : A new list of floats likelihoods containing the filtered values ."""

implementation:
redThreshold = 1.0 - redThreshold
yellowThreshold = 1.0 - yellowThreshold
# The first value is untouched
filteredLikelihoods = [ likelihoods [ 0 ] ]
for i , v in enumerate ( likelihoods [ 1 : ] ) :
    if v <= redThreshold : # Value is in the redzone
        if likelihoods [ i ] > redThreshold : # Previous value is not in redzone , so leave as - is
            filteredLikelihoods . append ( v )
        else :
            filteredLikelihoods . append ( yellowThreshold )
    else : # Value is below the redzone , so leave as - is
        filteredLikelihoods . append ( v )
return filteredLikelihoods


signature:
def _anomalyScoreMovingAverage ( anomalyScores , windowSize = 10 , verbosity = 0 , ) :
    """Given a list of anomaly scores return a list of averaged records .
 anomalyScores is assumed to be a list of records of the form :
 [ datetime . datetime ( 2013 , 8 , 10 , 23 , 0 ) , 6.0 , 1.0]
 Each record in the returned list list contains :
 [ datetime , value , averagedScore ]
 * Note : * we only average the anomaly score ."""

implementation:
historicalValues = [ ]
total = 0.0
averagedRecordList = [ ]
# Aggregated records
for record in anomalyScores : # Skip ( but log ) records without correct number of entries
    if not isinstance ( record , ( list , tuple ) ) or len ( record ) != 3 :
        if verbosity >= 1 :
            print ( "Malformed record:" , record )
        continue
    avg , historicalValues , total = ( MovingAverage . compute ( historicalValues , total , record [ 2 ] , windowSize ) )
    averagedRecordList . append ( [ record [ 0 ] , record [ 1 ] , avg ] )
    if verbosity > 2 :
        print ( "Aggregating input record:" , record )
        print ( "Result:" , [ record [ 0 ] , record [ 1 ] , avg ] )
return averagedRecordList , historicalValues , total


signature:
def estimateNormal ( sampleData , performLowerBoundCheck = True ) :
    """: param sampleData :
 : type sampleData : Numpy array .
 : param performLowerBoundCheck :
 : type performLowerBoundCheck : bool
 : returns : A dict containing the parameters of a normal distribution based on
 the ` ` sampleData ` ` ."""

implementation:
params = { "name" : "normal" , "mean" : numpy . mean ( sampleData ) , "variance" : numpy . var ( sampleData ) , }
if performLowerBoundCheck : # Handle edge case of almost no deviations and super low anomaly scores . We
# find that such low anomaly means can happen , but then the slightest blip
# of anomaly score can cause the likelihood to jump up to red .
    if params [ "mean" ] < 0.03 :
        params [ "mean" ] = 0.03
    # Catch all for super low variance to handle numerical precision issues
    if params [ "variance" ] < 0.0003 :
        params [ "variance" ] = 0.0003
# Compute standard deviation
if params [ "variance" ] > 0 :
    params [ "stdev" ] = math . sqrt ( params [ "variance" ] )
else :
    params [ "stdev" ] = 0
return params


signature:
def tailProbability ( x , distributionParams ) :
    """Given the normal distribution specified by the mean and standard deviation
 in distributionParams , return the probability of getting samples further
 from the mean . For values above the mean , this is the probability of getting
 samples > x and for values below the mean , the probability of getting
 samples < x . This is the Q - function : the tail probability of the normal distribution .
 : param distributionParams : dict with ' mean ' and ' stdev ' of the distribution"""

implementation:
if "mean" not in distributionParams or "stdev" not in distributionParams :
    raise RuntimeError ( "Insufficient parameters to specify the distribution." )
if x < distributionParams [ "mean" ] : # Gaussian is symmetrical around mean , so flip to get the tail probability
    xp = 2 * distributionParams [ "mean" ] - x
    return tailProbability ( xp , distributionParams )
# Calculate the Q function with the complementary error function , explained
# here : http : / / www . gaussianwaves . com / 2012/07 / q - function - and - error - functions
z = ( x - distributionParams [ "mean" ] ) / distributionParams [ "stdev" ]
return 0.5 * math . erfc ( z / 1.4142 )


signature:
def isValidEstimatorParams ( p ) :
    """: returns : ` ` True ` ` if ` ` p ` ` is a valid estimator params as might be returned
 by ` ` estimateAnomalyLikelihoods ( ) ` ` or ` ` updateAnomalyLikelihoods ` ` ,
 ` ` False ` ` otherwise . Just does some basic validation ."""

implementation:
if not isinstance ( p , dict ) :
    return False
if "distribution" not in p :
    return False
if "movingAverage" not in p :
    return False
dist = p [ "distribution" ]
if not ( "mean" in dist and "name" in dist and "variance" in dist and "stdev" in dist ) :
    return False
return True


signature:
def _calcSkipRecords ( numIngested , windowSize , learningPeriod ) :
    """Return the value of skipRecords for passing to estimateAnomalyLikelihoods
 If ` windowSize ` is very large ( bigger than the amount of data ) then this
 could just return ` learningPeriod ` . But when some values have fallen out of
 the historical sliding window of anomaly records , then we have to take those
 into account as well so we return the ` learningPeriod ` minus the number
 shifted out .
 : param numIngested - ( int ) number of data points that have been added to the
 sliding window of historical data points .
 : param windowSize - ( int ) size of sliding window of historical data points .
 : param learningPeriod - ( int ) the number of iterations required for the
 algorithm to learn the basic patterns in the dataset and for the anomaly
 score to ' settle down ' ."""

implementation:
numShiftedOut = max ( 0 , numIngested - windowSize )
return min ( numIngested , max ( 0 , learningPeriod - numShiftedOut ) )


signature:
def read ( cls , proto ) :
    """capnp deserialization method for the anomaly likelihood object
 : param proto : ( Object ) capnp proto object specified in
 nupic . regions . anomaly _ likelihood . capnp
 : returns : ( Object ) the deserialized AnomalyLikelihood object"""

implementation:
# pylint : disable = W0212
anomalyLikelihood = object . __new__ ( cls )
anomalyLikelihood . _iteration = proto . iteration
anomalyLikelihood . _historicalScores = collections . deque ( maxlen = proto . historicWindowSize )
for i , score in enumerate ( proto . historicalScores ) :
    anomalyLikelihood . _historicalScores . append ( ( i , score . value , score . anomalyScore ) )
if proto . distribution . name : # is " " when there is no distribution .
    anomalyLikelihood . _distribution = dict ( )
    anomalyLikelihood . _distribution [ 'distribution' ] = dict ( )
    anomalyLikelihood . _distribution [ 'distribution' ] [ "name" ] = proto . distribution . name
    anomalyLikelihood . _distribution [ 'distribution' ] [ "mean" ] = proto . distribution . mean
    anomalyLikelihood . _distribution [ 'distribution' ] [ "variance" ] = proto . distribution . variance
    anomalyLikelihood . _distribution [ 'distribution' ] [ "stdev" ] = proto . distribution . stdev
    anomalyLikelihood . _distribution [ "movingAverage" ] = { }
    anomalyLikelihood . _distribution [ "movingAverage" ] [ "windowSize" ] = proto . distribution . movingAverage . windowSize
    anomalyLikelihood . _distribution [ "movingAverage" ] [ "historicalValues" ] = [ ]
    for value in proto . distribution . movingAverage . historicalValues :
        anomalyLikelihood . _distribution [ "movingAverage" ] [ "historicalValues" ] . append ( value )
    anomalyLikelihood . _distribution [ "movingAverage" ] [ "total" ] = proto . distribution . movingAverage . total
    anomalyLikelihood . _distribution [ "historicalLikelihoods" ] = [ ]
    for likelihood in proto . distribution . historicalLikelihoods :
        anomalyLikelihood . _distribution [ "historicalLikelihoods" ] . append ( likelihood )
else :
    anomalyLikelihood . _distribution = None
anomalyLikelihood . _probationaryPeriod = proto . probationaryPeriod
anomalyLikelihood . _learningPeriod = proto . learningPeriod
anomalyLikelihood . _reestimationPeriod = proto . reestimationPeriod
# pylint : enable = W0212
return anomalyLikelihood


signature:
def write ( self , proto ) :
    """capnp serialization method for the anomaly likelihood object
 : param proto : ( Object ) capnp proto object specified in
 nupic . regions . anomaly _ likelihood . capnp"""

implementation:
proto . iteration = self . _iteration
pHistScores = proto . init ( 'historicalScores' , len ( self . _historicalScores ) )
for i , score in enumerate ( list ( self . _historicalScores ) ) :
    _ , value , anomalyScore = score
    record = pHistScores [ i ]
    record . value = float ( value )
    record . anomalyScore = float ( anomalyScore )
if self . _distribution :
    proto . distribution . name = self . _distribution [ "distribution" ] [ "name" ]
    proto . distribution . mean = float ( self . _distribution [ "distribution" ] [ "mean" ] )
    proto . distribution . variance = float ( self . _distribution [ "distribution" ] [ "variance" ] )
    proto . distribution . stdev = float ( self . _distribution [ "distribution" ] [ "stdev" ] )
    proto . distribution . movingAverage . windowSize = float ( self . _distribution [ "movingAverage" ] [ "windowSize" ] )
    historicalValues = self . _distribution [ "movingAverage" ] [ "historicalValues" ]
    pHistValues = proto . distribution . movingAverage . init ( "historicalValues" , len ( historicalValues ) )
    for i , value in enumerate ( historicalValues ) :
        pHistValues [ i ] = float ( value )
    # proto . distribution . movingAverage . historicalValues = self . _ distribution [ " movingAverage " ] [ " historicalValues " ]
    proto . distribution . movingAverage . total = float ( self . _distribution [ "movingAverage" ] [ "total" ] )
    historicalLikelihoods = self . _distribution [ "historicalLikelihoods" ]
    pHistLikelihoods = proto . distribution . init ( "historicalLikelihoods" , len ( historicalLikelihoods ) )
    for i , likelihood in enumerate ( historicalLikelihoods ) :
        pHistLikelihoods [ i ] = float ( likelihood )
proto . probationaryPeriod = self . _probationaryPeriod
proto . learningPeriod = self . _learningPeriod
proto . reestimationPeriod = self . _reestimationPeriod
proto . historicWindowSize = self . _historicalScores . maxlen


signature:
def anomalyProbability ( self , value , anomalyScore , timestamp = None ) :
    """Compute the probability that the current value plus anomaly score represents
 an anomaly given the historical distribution of anomaly scores . The closer
 the number is to 1 , the higher the chance it is an anomaly .
 : param value : the current metric ( " raw " ) input value , eg . " orange " , or
 '21.2 ' ( deg . Celsius ) , . . .
 : param anomalyScore : the current anomaly score
 : param timestamp : [ optional ] timestamp of the ocurrence ,
 default ( None ) results in using iteration step .
 : returns : the anomalyLikelihood for this record ."""

implementation:
if timestamp is None :
    timestamp = self . _iteration
dataPoint = ( timestamp , value , anomalyScore )
# We ignore the first probationaryPeriod data points
if self . _iteration < self . _probationaryPeriod :
    likelihood = 0.5
else : # On a rolling basis we re - estimate the distribution
    if ( ( self . _distribution is None ) or ( self . _iteration % self . _reestimationPeriod == 0 ) ) :
        numSkipRecords = self . _calcSkipRecords ( numIngested = self . _iteration , windowSize = self . _historicalScores . maxlen , learningPeriod = self . _learningPeriod )
        _ , _ , self . _distribution = estimateAnomalyLikelihoods ( self . _historicalScores , skipRecords = numSkipRecords )
    likelihoods , _ , self . _distribution = updateAnomalyLikelihoods ( [ dataPoint ] , self . _distribution )
    likelihood = 1.0 - likelihoods [ 0 ]
# Before we exit update historical scores and iteration
self . _historicalScores . append ( dataPoint )
self . _iteration += 1
return likelihood


signature:
def _getImpl ( self , model ) :
    """Creates and returns the _ IterationPhase - based instance corresponding
 to this phase specification
 model : Model instance"""

implementation:
impl = _IterationPhaseLearnOnly ( model = model , nIters = self . __nIters )
return impl


implementation:
impl = _IterationPhaseInferOnly ( model = model , nIters = self . __nIters , inferenceArgs = self . __inferenceArgs )
return impl


signature:
def replaceIterationCycle ( self , phaseSpecs ) :
    """Replaces the Iteration Cycle phases
 : param phaseSpecs : Iteration cycle description consisting of a sequence of
 IterationPhaseSpecXXXXX elements that are performed in the
 given order"""

implementation:
# Replace our phase manager
self . __phaseManager = _PhaseManager ( model = self . __model , phaseSpecs = phaseSpecs )
return


signature:
def handleInputRecord ( self , inputRecord ) :
    """Processes the given record according to the current iteration cycle phase
 : param inputRecord : ( object ) record expected to be returned from
 : meth : ` nupic . data . record _ stream . RecordStreamIface . getNextRecord ` .
 : returns : : class : ` nupic . frameworks . opf . opf _ utils . ModelResult `"""

implementation:
assert inputRecord , "Invalid inputRecord: %r" % inputRecord
results = self . __phaseManager . handleInputRecord ( inputRecord )
metrics = self . __metricsMgr . update ( results )
# Execute task - postIter callbacks
for cb in self . __userCallbacks [ 'postIter' ] :
    cb ( self . __model )
results . metrics = metrics
# Return the input and predictions for this record
return results


signature:
def __advancePhase ( self ) :
    """Advance to the next iteration cycle phase"""

implementation:
self . __currentPhase = self . __phaseCycler . next ( )
self . __currentPhase . enterPhase ( )
return


signature:
def handleInputRecord ( self , inputRecord ) :
    """Processes the given record according to the current phase
 inputRecord : record object formatted according to
 nupic . data . FileSource . getNext ( ) result format .
 Returns : An opf _ utils . ModelResult object with the inputs and inferences
 after the current record is processed by the model"""

implementation:
results = self . __model . run ( inputRecord )
shouldContinue = self . __currentPhase . advance ( )
if not shouldContinue :
    self . __advancePhase ( )
return results


signature:
def enterPhase ( self ) :
    """Performs initialization that is necessary upon entry to the phase . Must
 be called before handleInputRecord ( ) at the beginning of each phase"""

implementation:
self . __iter = iter ( xrange ( self . __nIters ) )
# Prime the iterator
self . __iter . next ( )


signature:
def advance ( self ) :
    """Advances the iteration ;
 Returns : True if more iterations remain ; False if this is the final
 iteration ."""

implementation:
hasMore = True
try :
    self . __iter . next ( )
except StopIteration :
    self . __iter = None
    hasMore = False
return hasMore


signature:
def enterPhase ( self ) :
    """[ _ IterationPhase method implementation ]
 Performs initialization that is necessary upon entry to the phase . Must
 be called before handleInputRecord ( ) at the beginning of each phase"""

implementation:
super ( _IterationPhaseLearnOnly , self ) . enterPhase ( )
self . __model . enableLearning ( )
self . __model . disableInference ( )
return


implementation:
super ( _IterationPhaseInferCommon , self ) . enterPhase ( )
self . _model . enableInference ( inferenceArgs = self . _inferenceArgs )
return


signature:
def write ( self , proto ) :
    """Serialize via capnp
 : param proto : capnp PreviousValueModelProto message builder"""

implementation:
super ( PreviousValueModel , self ) . writeBaseToProto ( proto . modelBase )
proto . fieldNames = self . _fieldNames
proto . fieldTypes = self . _fieldTypes
if self . _predictedField :
    proto . predictedField = self . _predictedField
proto . predictionSteps = self . _predictionSteps


signature:
def read ( cls , proto ) :
    """Deserialize via capnp
 : param proto : capnp PreviousValueModelProto message reader
 : returns : new instance of PreviousValueModel deserialized from the given
 proto"""

implementation:
instance = object . __new__ ( cls )
super ( PreviousValueModel , instance ) . __init__ ( proto = proto . modelBase )
instance . _logger = opf_utils . initLogger ( instance )
if len ( proto . predictedField ) :
    instance . _predictedField = proto . predictedField
else :
    instance . _predictedField = None
instance . _fieldNames = list ( proto . fieldNames )
instance . _fieldTypes = list ( proto . fieldTypes )
instance . _predictionSteps = list ( proto . predictionSteps )
return instance


signature:
def lscsum ( lx , epsilon = None ) :
    """Accepts log - values as input , exponentiates them , computes the sum ,
 then converts the sum back to log - space and returns the result .
 Handles underflow by rescaling so that the largest values is exactly 1.0."""

implementation:
lx = numpy . asarray ( lx )
base = lx . max ( )
# If the input is the log of 0 ' s , catch this condition before we generate
# an exception , and return the log ( 0)
if numpy . isinf ( base ) :
    return base
# If the user specified an epsilon and we are below it , return epsilon
if ( epsilon is not None ) and ( base < epsilon ) :
    return epsilon
x = numpy . exp ( lx - base )
ssum = x . sum ( )
result = numpy . log ( ssum ) + base
# try :
# conventional = numpy . log ( numpy . exp ( lx ) . sum ( ) )
# if not similar ( result , conventional ) :
# if numpy . isinf ( conventional ) . any ( ) and not numpy . isinf ( result ) . any ( ) :
# # print " Scaled log sum avoided underflow or overflow . "
# pass
# else :
# import sys
# print > > sys . stderr , " Warning : scaled log sum did not match . "
# print > > sys . stderr , " Scaled log result : "
# print > > sys . stderr , result
# print > > sys . stderr , " Conventional result : "
# print > > sys . stderr , conventional
# except FloatingPointError , e :
# # print " Scaled log sum avoided underflow or overflow . "
# pass
return result


signature:
def lscsum0 ( lx ) :
    """Accepts log - values as input , exponentiates them , sums down the rows
 ( first dimension ) , then converts the sum back to log - space and returns the result .
 Handles underflow by rescaling so that the largest values is exactly 1.0."""

implementation:
# rows = lx . shape [ 0]
# columns = numpy . prod ( lx . shape [ 1 : ] )
# lx = lx . reshape ( rows , columns )
# bases = lx . max ( 1 ) . reshape ( rows , 1)
# bases = lx . max ( 0 ) . reshape ( ( 1 , ) + lx . shape [ 1 : ] )
lx = numpy . asarray ( lx )
bases = lx . max ( 0 )
# Don ' t need to reshape in the case of 0.
x = numpy . exp ( lx - bases )
ssum = x . sum ( 0 )
result = numpy . log ( ssum ) + bases
try :
    conventional = numpy . log ( numpy . exp ( lx ) . sum ( 0 ) )
    if not similar ( result , conventional ) :
        if numpy . isinf ( conventional ) . any ( ) and not numpy . isinf ( result ) . any ( ) : # print " Scaled log sum down axis 0 avoided underflow or overflow . "
            pass
        else :
            import sys
            print >> sys . stderr , "Warning: scaled log sum down axis 0 did not match."
            print >> sys . stderr , "Scaled log result:"
            print >> sys . stderr , result
            print >> sys . stderr , "Conventional result:"
            print >> sys . stderr , conventional
except FloatingPointError , e : # print " Scaled log sum down axis 0 avoided underflow or overflow . "
    pass
return result


signature:
def normalize ( lx ) :
    """Accepts log - values as input , exponentiates them ,
 normalizes and returns the result .
 Handles underflow by rescaling so that the largest values is exactly 1.0."""

implementation:
lx = numpy . asarray ( lx )
base = lx . max ( )
x = numpy . exp ( lx - base )
result = x / x . sum ( )
conventional = ( numpy . exp ( lx ) / numpy . exp ( lx ) . sum ( ) )
assert similar ( result , conventional )
return result


signature:
def nsum0 ( lx ) :
    """Accepts log - values as input , exponentiates them , sums down the rows
 ( first dimension ) , normalizes and returns the result .
 Handles underflow by rescaling so that the largest values is exactly 1.0."""

implementation:
lx = numpy . asarray ( lx )
base = lx . max ( )
x = numpy . exp ( lx - base )
ssum = x . sum ( 0 )
result = ssum / ssum . sum ( )
conventional = ( numpy . exp ( lx ) . sum ( 0 ) / numpy . exp ( lx ) . sum ( ) )
assert similar ( result , conventional )
return result


signature:
def logSumExp ( A , B , out = None ) :
    """returns log ( exp ( A ) + exp ( B ) ) . A and B are numpy arrays"""

implementation:
if out is None :
    out = numpy . zeros ( A . shape )
indicator1 = A >= B
indicator2 = numpy . logical_not ( indicator1 )
out [ indicator1 ] = A [ indicator1 ] + numpy . log1p ( numpy . exp ( B [ indicator1 ] - A [ indicator1 ] ) )
out [ indicator2 ] = B [ indicator2 ] + numpy . log1p ( numpy . exp ( A [ indicator2 ] - B [ indicator2 ] ) )
return out


signature:
def logDiffExp ( A , B , out = None ) :
    """returns log ( exp ( A ) - exp ( B ) ) . A and B are numpy arrays . values in A should be
 greater than or equal to corresponding values in B"""

implementation:
if out is None :
    out = numpy . zeros ( A . shape )
indicator1 = A >= B
assert indicator1 . all ( ) , "Values in the first array should be greater than the values in the second"
out [ indicator1 ] = A [ indicator1 ] + numpy . log ( 1 - numpy . exp ( B [ indicator1 ] - A [ indicator1 ] ) )
return out


signature:
def _getFieldIndexBySpecial ( fields , special ) :
    """Return index of the field matching the field meta special value .
 : param fields : sequence of nupic . data . fieldmeta . FieldMetaInfo objects
 representing the fields of a stream
 : param special : one of the special field attribute values from
 nupic . data . fieldmeta . FieldMetaSpecial
 : returns : first zero - based index of the field tagged with the target field
 meta special attribute ; None if no such field"""

implementation:
for i , field in enumerate ( fields ) :
    if field . special == special :
        return i
return None


signature:
def encode ( self , inputRow ) :
    """Encodes the given input row as a dict , with the
 keys being the field names . This also adds in some meta fields :
 ' _ category ' : The value from the category field ( if any )
 ' _ reset ' : True if the reset field was True ( if any )
 ' _ sequenceId ' : the value from the sequenceId field ( if any )
 : param inputRow : sequence of values corresponding to a single input metric
 data row
 : rtype : dict"""

implementation:
# Create the return dict
result = dict ( zip ( self . _fieldNames , inputRow ) )
# Add in the special fields
if self . _categoryFieldIndex is not None : # category value can be an int or a list
    if isinstance ( inputRow [ self . _categoryFieldIndex ] , int ) :
        result [ '_category' ] = [ inputRow [ self . _categoryFieldIndex ] ]
    else :
        result [ '_category' ] = ( inputRow [ self . _categoryFieldIndex ] if inputRow [ self . _categoryFieldIndex ] else [ None ] )
else :
    result [ '_category' ] = [ None ]
if self . _resetFieldIndex is not None :
    result [ '_reset' ] = int ( bool ( inputRow [ self . _resetFieldIndex ] ) )
else :
    result [ '_reset' ] = 0
if self . _learningFieldIndex is not None :
    result [ '_learning' ] = int ( bool ( inputRow [ self . _learningFieldIndex ] ) )
result [ '_timestampRecordIdx' ] = None
if self . _timestampFieldIndex is not None :
    result [ '_timestamp' ] = inputRow [ self . _timestampFieldIndex ]
    # Compute the record index based on timestamp
    result [ '_timestampRecordIdx' ] = self . _computeTimestampRecordIdx ( inputRow [ self . _timestampFieldIndex ] )
else :
    result [ '_timestamp' ] = None
# Figure out the sequence ID
hasReset = self . _resetFieldIndex is not None
hasSequenceId = self . _sequenceFieldIndex is not None
if hasReset and not hasSequenceId : # Reset only
    if result [ '_reset' ] :
        self . _sequenceId += 1
    sequenceId = self . _sequenceId
elif not hasReset and hasSequenceId :
    sequenceId = inputRow [ self . _sequenceFieldIndex ]
    result [ '_reset' ] = int ( sequenceId != self . _sequenceId )
    self . _sequenceId = sequenceId
elif hasReset and hasSequenceId :
    sequenceId = inputRow [ self . _sequenceFieldIndex ]
else :
    sequenceId = 0
if sequenceId is not None :
    result [ '_sequenceId' ] = hash ( sequenceId )
else :
    result [ '_sequenceId' ] = None
return result


signature:
def _computeTimestampRecordIdx ( self , recordTS ) :
    """Give the timestamp of a record ( a datetime object ) , compute the record ' s
 timestamp index - this is the timestamp divided by the aggregation period .
 Parameters :
 recordTS : datetime instance
 retval : record timestamp index , or None if no aggregation period"""

implementation:
if self . _aggregationPeriod is None :
    return None
# Base record index on number of elapsed months if aggregation is in
# months
if self . _aggregationPeriod [ 'months' ] > 0 :
    assert self . _aggregationPeriod [ 'seconds' ] == 0
    result = int ( ( recordTS . year * 12 + ( recordTS . month - 1 ) ) / self . _aggregationPeriod [ 'months' ] )
# Base record index on elapsed seconds
elif self . _aggregationPeriod [ 'seconds' ] > 0 :
    delta = recordTS - datetime . datetime ( year = 1 , month = 1 , day = 1 )
    deltaSecs = delta . days * 24 * 60 * 60 + delta . seconds + delta . microseconds / 1000000.0
    result = int ( deltaSecs / self . _aggregationPeriod [ 'seconds' ] )
else :
    result = None
return result


signature:
def getNextRecordDict ( self ) :
    """Returns next available data record from the storage as a dict , with the
 keys being the field names . This also adds in some meta fields :
 - ` ` _ category ` ` : The value from the category field ( if any )
 - ` ` _ reset ` ` : True if the reset field was True ( if any )
 - ` ` _ sequenceId ` ` : the value from the sequenceId field ( if any )"""

implementation:
values = self . getNextRecord ( )
if values is None :
    return None
if not values :
    return dict ( )
if self . _modelRecordEncoder is None :
    self . _modelRecordEncoder = ModelRecordEncoder ( fields = self . getFields ( ) , aggregationPeriod = self . getAggregationMonthsAndSeconds ( ) )
return self . _modelRecordEncoder . encode ( values )


signature:
def getFieldMin ( self , fieldName ) :
    """If underlying implementation does not support min / max stats collection ,
 or if a field type does not support min / max ( non scalars ) , the return
 value will be None .
 : param fieldName : ( string ) name of field to get min
 : returns : current minimum value for the field ` ` fieldName ` ` ."""

implementation:
stats = self . getStats ( )
if stats == None :
    return None
minValues = stats . get ( 'min' , None )
if minValues == None :
    return None
index = self . getFieldNames ( ) . index ( fieldName )
return minValues [ index ]


signature:
def getFieldMax ( self , fieldName ) :
    """If underlying implementation does not support min / max stats collection ,
 or if a field type does not support min / max ( non scalars ) , the return
 value will be None .
 : param fieldName : ( string ) name of field to get max
 : returns : current maximum value for the field ` ` fieldName ` ` ."""

implementation:
stats = self . getStats ( )
if stats == None :
    return None
maxValues = stats . get ( 'max' , None )
if maxValues == None :
    return None
index = self . getFieldNames ( ) . index ( fieldName )
return maxValues [ index ]


signature:
def debug ( self , msg , * args , ** kwargs ) :
    """Log ' msg % args ' with severity ' DEBUG ' .
 To pass exception information , use the keyword argument exc _ info with
 a true value , e . g .
 logger . debug ( " Houston , we have a % s " , " thorny problem " , exc _ info = 1)"""

implementation: self . _baseLogger . debug ( self , self . getExtendedMsg ( msg ) , * args , ** kwargs )


signature:
def info ( self , msg , * args , ** kwargs ) :
    """Log ' msg % args ' with severity ' INFO ' .
 To pass exception information , use the keyword argument exc _ info with
 a true value , e . g .
 logger . info ( " Houston , we have a % s " , " interesting problem " , exc _ info = 1)"""

implementation: self . _baseLogger . info ( self , self . getExtendedMsg ( msg ) , * args , ** kwargs )


signature:
def warning ( self , msg , * args , ** kwargs ) :
    """Log ' msg % args ' with severity ' WARNING ' .
 To pass exception information , use the keyword argument exc _ info with
 a true value , e . g .
 logger . warning ( " Houston , we have a % s " , " bit of a problem " , exc _ info = 1)"""

implementation: self . _baseLogger . warning ( self , self . getExtendedMsg ( msg ) , * args , ** kwargs )


signature:
def error ( self , msg , * args , ** kwargs ) :
    """Log ' msg % args ' with severity ' ERROR ' .
 To pass exception information , use the keyword argument exc _ info with
 a true value , e . g .
 logger . error ( " Houston , we have a % s " , " major problem " , exc _ info = 1)"""

implementation: self . _baseLogger . error ( self , self . getExtendedMsg ( msg ) , * args , ** kwargs )


signature:
def critical ( self , msg , * args , ** kwargs ) :
    """Log ' msg % args ' with severity ' CRITICAL ' .
 To pass exception information , use the keyword argument exc _ info with
 a true value , e . g .
 logger . critical ( " Houston , we have a % s " , " major disaster " , exc _ info = 1)"""

implementation: self . _baseLogger . critical ( self , self . getExtendedMsg ( msg ) , * args , ** kwargs )


signature:
def log ( self , level , msg , * args , ** kwargs ) :
    """Log ' msg % args ' with the integer severity ' level ' .
 To pass exception information , use the keyword argument exc _ info with
 a true value , e . g .
 logger . log ( level , " We have a % s " , " mysterious problem " , exc _ info = 1)"""

implementation: self . _baseLogger . log ( self , level , self . getExtendedMsg ( msg ) , * args , ** kwargs )


signature:
def initFilter ( input , filterInfo = None ) :
    """Initializes internal filter variables for further processing .
 Returns a tuple ( function to call , parameters for the filter call )
 The filterInfo is a dict . Here is an example structure :
 { fieldName : { ' min ' : x ,
 ' max ' : y ,
 ' type ' : ' category ' , # or ' number '
 ' acceptValues ' : [ ' foo ' , ' bar ' ] ,
 This returns the following :
 ( filterFunc , ( ( fieldIdx , fieldFilterFunc , filterDict ) ,
 Where fieldIdx is the index of the field within each record
 fieldFilterFunc returns True if the value is " OK " ( within min , max or
 part of acceptValues )
 fieldDict is a dict containing ' type ' , ' min ' , max ' , ' acceptValues '"""

implementation:
if filterInfo is None :
    return None
# Build an array of index / func to call on record [ index ]
filterList = [ ]
for i , fieldName in enumerate ( input . getFieldNames ( ) ) :
    fieldFilter = filterInfo . get ( fieldName , None )
    if fieldFilter == None :
        continue
    var = dict ( )
    var [ 'acceptValues' ] = None
    min = fieldFilter . get ( 'min' , None )
    max = fieldFilter . get ( 'max' , None )
    var [ 'min' ] = min
    var [ 'max' ] = max
    if fieldFilter [ 'type' ] == 'category' :
        var [ 'acceptValues' ] = fieldFilter [ 'acceptValues' ]
        fp = lambda x : ( x [ 'value' ] != SENTINEL_VALUE_FOR_MISSING_DATA and x [ 'value' ] in x [ 'acceptValues' ] )
    elif fieldFilter [ 'type' ] == 'number' :
        if min != None and max != None :
            fp = lambda x : ( x [ 'value' ] != SENTINEL_VALUE_FOR_MISSING_DATA and x [ 'value' ] >= x [ 'min' ] and x [ 'value' ] <= x [ 'max' ] )
        elif min != None :
            fp = lambda x : ( x [ 'value' ] != SENTINEL_VALUE_FOR_MISSING_DATA and x [ 'value' ] >= x [ 'min' ] )
        else :
            fp = lambda x : ( x [ 'value' ] != SENTINEL_VALUE_FOR_MISSING_DATA and x [ 'value' ] <= x [ 'max' ] )
    filterList . append ( ( i , fp , var ) )
return ( _filterRecord , filterList )


signature:
def _filterRecord ( filterList , record ) :
    """Takes a record and returns true if record meets filter criteria ,
 false otherwise"""

implementation:
for ( fieldIdx , fp , params ) in filterList :
    x = dict ( )
    x [ 'value' ] = record [ fieldIdx ]
    x [ 'acceptValues' ] = params [ 'acceptValues' ]
    x [ 'min' ] = params [ 'min' ]
    x [ 'max' ] = params [ 'max' ]
    if not fp ( x ) :
        return False
# None of the field filters triggered , accept the record as a good one
return True


signature:
def _aggr_sum ( inList ) :
    """Returns sum of the elements in the list . Missing items are replaced with
 the mean value"""

implementation:
aggrMean = _aggr_mean ( inList )
if aggrMean == None :
    return None
aggrSum = 0
for elem in inList :
    if elem != SENTINEL_VALUE_FOR_MISSING_DATA :
        aggrSum += elem
    else :
        aggrSum += aggrMean
return aggrSum


signature:
def _aggr_mean ( inList ) :
    """Returns mean of non - None elements of the list"""

implementation:
aggrSum = 0
nonNone = 0
for elem in inList :
    if elem != SENTINEL_VALUE_FOR_MISSING_DATA :
        aggrSum += elem
        nonNone += 1
if nonNone != 0 :
    return aggrSum / nonNone
else :
    return None


signature:
def _aggr_mode ( inList ) :
    """Returns most common value seen in the non - None elements of the list"""

implementation:
valueCounts = dict ( )
nonNone = 0
for elem in inList :
    if elem == SENTINEL_VALUE_FOR_MISSING_DATA :
        continue
    nonNone += 1
    if elem in valueCounts :
        valueCounts [ elem ] += 1
    else :
        valueCounts [ elem ] = 1
# Get the most common one
if nonNone == 0 :
    return None
# Sort by counts
sortedCounts = valueCounts . items ( )
sortedCounts . sort ( cmp = lambda x , y : x [ 1 ] - y [ 1 ] , reverse = True )
return sortedCounts [ 0 ] [ 0 ]


signature:
def _aggr_weighted_mean ( inList , params ) :
    """Weighted mean uses params ( must be the same size as inList ) and
 makes weighed mean of inList"""

implementation:
assert ( len ( inList ) == len ( params ) )
# If all weights are 0 , then the value is not defined , return None ( missing )
weightsSum = sum ( params )
if weightsSum == 0 :
    return None
weightedMean = 0
for i , elem in enumerate ( inList ) :
    weightedMean += elem * params [ i ]
return weightedMean / weightsSum


signature:
def generateDataset ( aggregationInfo , inputFilename , outputFilename = None ) :
    """Generate a dataset of aggregated values
 Parameters :
 aggregationInfo : a dictionary that contains the following entries
 - fields : a list of pairs . Each pair is a field name and an
 aggregation function ( e . g . sum ) . The function will be used to aggregate
 multiple values during the aggregation period .
 aggregation period : 0 or more of unit = value fields ; allowed units are :
 [ years months ] |
 [ weeks days hours minutes seconds milliseconds microseconds ]
 NOTE : years and months are mutually - exclusive with the other units .
 See getEndTime ( ) and _ aggregate ( ) for more details .
 Example1 : years = 1 , months = 6,
 Example2 : hours = 1 , minutes = 30,
 If none of the period fields are specified or if all that are specified
 have values of 0 , then aggregation will be suppressed , and the given
 inputFile parameter value will be returned .
 inputFilename : filename of the input dataset within examples / prediction / data
 outputFilename : name for the output file . If not given , a name will be
 generated based on the input filename and the aggregation params
 retval : Name of the generated output file . This will be the same as the input
 file name if no aggregation needed to be performed
 If the input file contained a time field , sequence id field or reset field
 that were not specified in aggregationInfo fields , those fields will be
 added automatically with the following rules :
 1 . The order will be R , S , T , rest of the fields
 2 . The aggregation function for all will be to pick the first : lambda x : x [ 0]
 Returns : the path of the aggregated data file if aggregation was performed
 ( in the same directory as the given input file ) ; if aggregation did not
 need to be performed , then the given inputFile argument value is returned ."""

implementation:
# Create the input stream
inputFullPath = resource_filename ( "nupic.datafiles" , inputFilename )
inputObj = FileRecordStream ( inputFullPath )
# Instantiate the aggregator
aggregator = Aggregator ( aggregationInfo = aggregationInfo , inputFields = inputObj . getFields ( ) )
# Is it a null aggregation ? If so , just return the input file unmodified
if aggregator . isNullAggregation ( ) :
    return inputFullPath
# If we were not given an output filename , create one based on the
# aggregation settings
if outputFilename is None :
    outputFilename = 'agg_%s' % os . path . splitext ( os . path . basename ( inputFullPath ) ) [ 0 ]
    timePeriods = 'years months weeks days ' 'hours minutes seconds milliseconds microseconds'
    for k in timePeriods . split ( ) :
        if aggregationInfo . get ( k , 0 ) > 0 :
            outputFilename += '_%s_%d' % ( k , aggregationInfo [ k ] )
    outputFilename += '.csv'
    outputFilename = os . path . join ( os . path . dirname ( inputFullPath ) , outputFilename )
# If some other process already started creating this file , simply
# wait for it to finish and return without doing anything
lockFilePath = outputFilename + '.please_wait'
if os . path . isfile ( outputFilename ) or os . path . isfile ( lockFilePath ) :
    while os . path . isfile ( lockFilePath ) :
        print 'Waiting for %s to be fully written by another process' % lockFilePath
        time . sleep ( 1 )
    return outputFilename
# Create the lock file
lockFD = open ( lockFilePath , 'w' )
# Create the output stream
outputObj = FileRecordStream ( streamID = outputFilename , write = True , fields = inputObj . getFields ( ) )
# Write all aggregated records to the output
while True :
    inRecord = inputObj . getNextRecord ( )
    ( aggRecord , aggBookmark ) = aggregator . next ( inRecord , None )
    if aggRecord is None and inRecord is None :
        break
    if aggRecord is not None :
        outputObj . appendRecord ( aggRecord )
return outputFilename


signature:
def getFilename ( aggregationInfo , inputFile ) :
    """Generate the filename for aggregated dataset
 The filename is based on the input filename and the
 aggregation period .
 Returns the inputFile if no aggregation required ( aggregation
 info has all 0 ' s )"""

implementation:
# Find the actual file , with an absolute path
inputFile = resource_filename ( "nupic.datafiles" , inputFile )
a = defaultdict ( lambda : 0 , aggregationInfo )
outputDir = os . path . dirname ( inputFile )
outputFile = 'agg_%s' % os . path . splitext ( os . path . basename ( inputFile ) ) [ 0 ]
noAggregation = True
timePeriods = 'years months weeks days ' 'hours minutes seconds milliseconds microseconds'
for k in timePeriods . split ( ) :
    if a [ k ] > 0 :
        noAggregation = False
        outputFile += '_%s_%d' % ( k , a [ k ] )
if noAggregation :
    return inputFile
outputFile += '.csv'
outputFile = os . path . join ( outputDir , outputFile )
return outputFile


signature:
def _getEndTime ( self , t ) :
    """Add the aggregation period to the input time t and return a datetime object
 Years and months are handled as aspecial case due to leap years
 and months with different number of dates . They can ' t be converted
 to a strict timedelta because a period of 3 months will have different
 durations actually . The solution is to just add the years and months
 fields directly to the current time .
 Other periods are converted to timedelta and just added to current time ."""

implementation:
assert isinstance ( t , datetime . datetime )
if self . _aggTimeDelta :
    return t + self . _aggTimeDelta
else :
    year = t . year + self . _aggYears + ( t . month - 1 + self . _aggMonths ) / 12
    month = ( t . month - 1 + self . _aggMonths ) % 12 + 1
    return t . replace ( year = year , month = month )


signature:
def _getFuncPtrAndParams ( self , funcName ) :
    """Given the name of an aggregation function , returns the function pointer
 and param .
 Parameters :
 funcName : a string ( name of function ) or funcPtr
 retval : ( funcPtr , param )"""

implementation:
params = None
if isinstance ( funcName , basestring ) :
    if funcName == 'sum' :
        fp = _aggr_sum
    elif funcName == 'first' :
        fp = _aggr_first
    elif funcName == 'last' :
        fp = _aggr_last
    elif funcName == 'mean' :
        fp = _aggr_mean
    elif funcName == 'max' :
        fp = max
    elif funcName == 'min' :
        fp = min
    elif funcName == 'mode' :
        fp = _aggr_mode
    elif funcName . startswith ( 'wmean:' ) :
        fp = _aggr_weighted_mean
        paramsName = funcName [ 6 : ]
        params = [ f [ 0 ] for f in self . _inputFields ] . index ( paramsName )
else :
    fp = funcName
return ( fp , params )


signature:
def _createAggregateRecord ( self ) :
    """Generate the aggregated output record
 Parameters :
 retval : outputRecord"""

implementation:
record = [ ]
for i , ( fieldIdx , aggFP , paramIdx ) in enumerate ( self . _fields ) :
    if aggFP is None : # this field is not supposed to be aggregated .
        continue
    values = self . _slice [ i ]
    refIndex = None
    if paramIdx is not None :
        record . append ( aggFP ( values , self . _slice [ paramIdx ] ) )
    else :
        record . append ( aggFP ( values ) )
return record


signature:
def next ( self , record , curInputBookmark ) :
    """Return the next aggregated record , if any
 Parameters :
 record : The input record ( values only ) from the input source , or
 None if the input has reached EOF ( this will cause this
 method to force completion of and return any partially
 aggregated time period )
 curInputBookmark : The bookmark to the next input record
 retval :
 ( outputRecord , inputBookmark )
 outputRecord : the aggregated record
 inputBookmark : a bookmark to the last position from the input that
 contributed to this aggregated record .
 If we don ' t have any aggregated records yet , returns ( None , None )
 The caller should generally do a loop like this :
 while True :
 inRecord = reader . getNextRecord ( )
 bookmark = reader . getBookmark ( )
 ( aggRecord , aggBookmark ) = aggregator . next ( inRecord , bookmark )
 # reached EOF ?
 if inRecord is None and aggRecord is None :
 break
 if aggRecord is not None :
 proessRecord ( aggRecord , aggBookmark )
 This method makes use of the self . _ slice member variable to build up
 the values we need to aggregate . This is a dict of lists . The keys are
 the field indices and the elements of each list are the values for that
 field . For example :
 self . _ siice = { 0 : [ 42 , 53 ] , 1 : [ 4.0 , 5.1 ] }"""

implementation:
# This will hold the aggregated record we return
outRecord = None
# This will hold the bookmark of the last input used within the
# aggregated record we return .
retInputBookmark = None
if record is not None : # Increment input count
    self . _inIdx += 1
    # print self . _ inIdx , record
    # Apply the filter , ignore the record if any field is unacceptable
    if self . _filter != None and not self . _filter [ 0 ] ( self . _filter [ 1 ] , record ) :
        return ( None , None )
    # If no aggregation info just return as - is
    if self . _nullAggregation :
        return ( record , curInputBookmark )
    # Do aggregation
    # Remember the very first record time stamp - it will be used as
    # the timestamp for all first records in all sequences to align
    # times for the aggregation / join of sequences .
    # For a set of aggregated records , it will use the beginning of the time
    # window as a timestamp for the set
    t = record [ self . _timeFieldIdx ]
    if self . _firstSequenceStartTime == None :
        self . _firstSequenceStartTime = t
    # Create initial startTime and endTime if needed
    if self . _startTime is None :
        self . _startTime = t
    if self . _endTime is None :
        self . _endTime = self . _getEndTime ( t )
        assert self . _endTime > t
    # print ' Processing line : ' , i , t , endTime
    # from dbgp . client import brk ; brk ( port = 9011)
    # Does this record have a reset signal or sequence Id associated with it ?
    # If so , see if we ' ve reached a sequence boundary
    if self . _resetFieldIdx is not None :
        resetSignal = record [ self . _resetFieldIdx ]
    else :
        resetSignal = None
    if self . _sequenceIdFieldIdx is not None :
        currSequenceId = record [ self . _sequenceIdFieldIdx ]
    else :
        currSequenceId = None
    newSequence = ( resetSignal == 1 and self . _inIdx > 0 ) or self . _sequenceId != currSequenceId or self . _inIdx == 0
    if newSequence :
        self . _sequenceId = currSequenceId
    # We end the aggregation chunk if we go past the end time
    # - OR - we get an out of order record ( t < startTime )
    sliceEnded = ( t >= self . _endTime or t < self . _startTime )
    # Time to generate a new output record ?
    if ( newSequence or sliceEnded ) and len ( self . _slice ) > 0 : # Create aggregated record
    # print ' Creating aggregate record . . . '
    # Make first record timestamp as the beginning of the time period ,
    # in case the first record wasn ' t falling on the beginning of the period
        for j , f in enumerate ( self . _fields ) :
            index = f [ 0 ]
            if index == self . _timeFieldIdx :
                self . _slice [ j ] [ 0 ] = self . _startTime
                break
        # Generate the aggregated record
        outRecord = self . _createAggregateRecord ( )
        retInputBookmark = self . _aggrInputBookmark
        # Reset the slice
        self . _slice = defaultdict ( list )
    # Add current record to slice ( Note keeping slices in memory ) . Each
    # field in the slice is a list of field values from all the sliced
    # records
    for j , f in enumerate ( self . _fields ) :
        index = f [ 0 ]
        # append the parsed field value to the proper aggregated slice field .
        self . _slice [ j ] . append ( record [ index ] )
        self . _aggrInputBookmark = curInputBookmark
    # If we ' ve encountered a new sequence , start aggregation over again
    if newSequence : # TODO : May use self . _ firstSequenceStartTime as a start for the new
    # sequence ( to align all sequences )
        self . _startTime = t
        self . _endTime = self . _getEndTime ( t )
    # If a slice just ended , re - compute the start and end time for the
    # next aggregated record
    if sliceEnded : # Did we receive an out of order record ? If so , go back and iterate
    # till we get to the next end time boundary .
        if t < self . _startTime :
            self . _endTime = self . _firstSequenceStartTime
        while t >= self . _endTime :
            self . _startTime = self . _endTime
            self . _endTime = self . _getEndTime ( self . _endTime )
    # If we have a record to return , do it now
    if outRecord is not None :
        return ( outRecord , retInputBookmark )
# Input reached EOF
# Aggregate one last time in the end if necessary
elif self . _slice : # Make first record timestamp as the beginning of the time period ,
# in case the first record wasn ' t falling on the beginning of the period
    for j , f in enumerate ( self . _fields ) :
        index = f [ 0 ]
        if index == self . _timeFieldIdx :
            self . _slice [ j ] [ 0 ] = self . _startTime
            break
    outRecord = self . _createAggregateRecord ( )
    retInputBookmark = self . _aggrInputBookmark
    self . _slice = defaultdict ( list )
# Return aggregated record
return ( outRecord , retInputBookmark )


signature:
def processClubAttendance ( f , clubs ) :
    """Process the attendance data of one club
 If the club already exists in the list update its data .
 If the club is new create a new Club object and add it to the dict
 The next step is to iterate over all the lines and add a record for each line .
 When reaching an empty line it means there are no more records for this club .
 Along the way some redundant lines are skipped . When the file ends the f . next ( )
 call raises a StopIteration exception and that ' s the sign to return False ,
 which indicates to the caller that there are no more clubs to process ."""

implementation:
try : # Skip as many empty lines as necessary ( file format inconsistent )
    line = f . next ( )
    while line == ',,,,,,,,,,,,,,,,,,,\n' :
        line = f . next ( )
    # The first non - empty line should have the name as the first field
    name = line . split ( ',' ) [ 0 ]
    # Create a new club object if needed
    if name not in clubs :
        clubs [ name ] = Club ( name )
    # Get the named club
    c = clubs [ name ]
    c . processAttendance ( f )
    return True
except StopIteration :
    return False


signature:
def processClubConsumption ( f , clubs ) :
    """Process the consumption a club
 - Skip the header line
 - Iterate over lines
 - Read 4 records at a time
 - Parse each line : club , date , time , consumption
 - Get club object from dictionary if needed
 - Aggregate consumption
 - Call club . processConsumption ( ) with data"""

implementation:
try : # Skip header line
    line = f . next ( )
    assert line . endswith ( '"   ","SITE_LOCATION_NAME","TIMESTAMP","TOTAL_KWH"\n' )
    valid_times = range ( 24 )
    t = 0
    # used to track time
    club = None
    clubName = None
    lastDate = None
    while True :
        assert t in valid_times
        consumption = 0
        for x in range ( 4 ) : # Read the line and get rid of the newline character
            line = f . next ( ) [ : - 1 ]
            fields = line . split ( ',' )
            assert len ( fields ) == 4
            for i , field in enumerate ( fields ) : # Strip the redundant double quotes
                assert field [ 0 ] == '"' and field [ - 1 ] == '"'
                fields [ i ] = field [ 1 : - 1 ]
            # Ignoring field 0 , which is just a running count
            # Get the club name
            name = fields [ 1 ]
            # Hack to fix inconsistent club names like : " Melbourne CBD - Melbourne Central " vs . " Melbourne Central "
            partialNames = ( 'Melbourne Central' , 'North Sydney' , 'Park St' , 'Pitt St' )
            for pn in partialNames :
                if pn in name :
                    name = pn
            # Locate the club if needed ( maybe )
            if name != clubName :
                clubName = name
                club = clubs [ name ]
            # Split the date ( time is counted using the t variable )
            tokens = fields [ 2 ] . split ( )
            # Verify that t = = 0 and consumption = = 0 when there is no time in the file
            if len ( tokens ) == 1 :
                assert consumption == 0 and t == 0
            # The first ( and sometimes only ) token is the date
            date = tokens [ 0 ]
            # Aggregate the consumption
            consumption += float ( fields [ 3 ] )
        # Update the Club object after aggregating the consumption of 4 lines
        club . updateRecord ( date , t , consumption )
        # Increment time
        t += 1
        t %= 24
except StopIteration :
    return


signature:
def run ( self , inputRecord ) :
    """Run one iteration of this model .
 : param inputRecord : ( object )
 A record object formatted according to
 : meth : ` ~ nupic . data . record _ stream . RecordStreamIface . getNextRecord ` or
 : meth : ` ~ nupic . data . record _ stream . RecordStreamIface . getNextRecordDict `
 result format .
 : returns : ( : class : ` ~ nupic . frameworks . opf . opf _ utils . ModelResult ` )
 An ModelResult namedtuple . The contents of ModelResult . inferences
 depends on the the specific inference type of this model , which
 can be queried by : meth : ` . getInferenceType ` ."""

implementation:
# 0 - based prediction index for ModelResult
predictionNumber = self . _numPredictions
self . _numPredictions += 1
result = opf_utils . ModelResult ( predictionNumber = predictionNumber , rawInput = inputRecord )
return result


signature:
def _getModelCheckpointFilePath ( checkpointDir ) :
    """Return the absolute path of the model ' s checkpoint file .
 : param checkpointDir : ( string )
 Directory of where the experiment is to be or was saved
 : returns : ( string ) An absolute path ."""

implementation:
path = os . path . join ( checkpointDir , "model.data" )
path = os . path . abspath ( path )
return path


signature:
def writeToCheckpoint ( self , checkpointDir ) :
    """Serializes model using capnproto and writes data to ` ` checkpointDir ` `"""

implementation:
proto = self . getSchema ( ) . new_message ( )
self . write ( proto )
checkpointPath = self . _getModelCheckpointFilePath ( checkpointDir )
# Clean up old saved state , if any
if os . path . exists ( checkpointDir ) :
    if not os . path . isdir ( checkpointDir ) :
        raise Exception ( ( "Existing filesystem entry <%s> is not a model" " checkpoint -- refusing to delete (not a directory)" ) % checkpointDir )
    if not os . path . isfile ( checkpointPath ) :
        raise Exception ( ( "Existing filesystem entry <%s> is not a model" " checkpoint -- refusing to delete" " (%s missing or not a file)" ) % ( checkpointDir , checkpointPath ) )
    shutil . rmtree ( checkpointDir )
# Create a new directory for saving state
self . __makeDirectoryFromAbsolutePath ( checkpointDir )
with open ( checkpointPath , 'wb' ) as f :
    proto . write ( f )


signature:
def readFromCheckpoint ( cls , checkpointDir ) :
    """Deserializes model from checkpointDir using capnproto"""

implementation:
checkpointPath = cls . _getModelCheckpointFilePath ( checkpointDir )
with open ( checkpointPath , 'r' ) as f :
    proto = cls . getSchema ( ) . read ( f , traversal_limit_in_words = _TRAVERSAL_LIMIT_IN_WORDS )
model = cls . read ( proto )
return model


signature:
def writeBaseToProto ( self , proto ) :
    """Save the state maintained by the Model base class
 : param proto : capnp ModelProto message builder"""

implementation:
inferenceType = self . getInferenceType ( )
# lower - case first letter to be compatible with capnproto enum naming
inferenceType = inferenceType [ : 1 ] . lower ( ) + inferenceType [ 1 : ]
proto . inferenceType = inferenceType
proto . numPredictions = self . _numPredictions
proto . learningEnabled = self . __learningEnabled
proto . inferenceEnabled = self . __inferenceEnabled
proto . inferenceArgs = json . dumps ( self . __inferenceArgs )


signature:
def save ( self , saveModelDir ) :
    """Save the model in the given directory .
 : param saveModelDir : ( string )
 Absolute directory path for saving the model . This directory should
 only be used to store a saved model . If the directory does not exist ,
 it will be created automatically and populated with model data . A
 pre - existing directory will only be accepted if it contains previously
 saved model data . If such a directory is given , the full contents of
 the directory will be deleted and replaced with current model data ."""

implementation:
logger = self . _getLogger ( )
logger . debug ( "(%s) Creating local checkpoint in %r..." , self , saveModelDir )
modelPickleFilePath = self . _getModelPickleFilePath ( saveModelDir )
# Clean up old saved state , if any
if os . path . exists ( saveModelDir ) :
    if not os . path . isdir ( saveModelDir ) :
        raise Exception ( ( "Existing filesystem entry <%s> is not a model" " checkpoint -- refusing to delete (not a directory)" ) % saveModelDir )
    if not os . path . isfile ( modelPickleFilePath ) :
        raise Exception ( ( "Existing filesystem entry <%s> is not a model" " checkpoint -- refusing to delete" " (%s missing or not a file)" ) % ( saveModelDir , modelPickleFilePath ) )
    shutil . rmtree ( saveModelDir )
# Create a new directory for saving state
self . __makeDirectoryFromAbsolutePath ( saveModelDir )
with open ( modelPickleFilePath , 'wb' ) as modelPickleFile :
    logger . debug ( "(%s) Pickling Model instance..." , self )
    pickle . dump ( self , modelPickleFile , protocol = pickle . HIGHEST_PROTOCOL )
    logger . debug ( "(%s) Finished pickling Model instance" , self )
# Tell the model to save extra data , if any , that ' s too big for pickling
self . _serializeExtraData ( extraDataDir = self . _getModelExtraDataDir ( saveModelDir ) )
logger . debug ( "(%s) Finished creating local checkpoint" , self )
return


signature:
def load ( cls , savedModelDir ) :
    """Load saved model .
 : param savedModelDir : ( string )
 Directory of where the experiment is to be or was saved
 : returns : ( : class : ` Model ` ) The loaded model instance"""

implementation:
logger = opf_utils . initLogger ( cls )
logger . debug ( "Loading model from local checkpoint at %r..." , savedModelDir )
# Load the model
modelPickleFilePath = Model . _getModelPickleFilePath ( savedModelDir )
with open ( modelPickleFilePath , 'rb' ) as modelPickleFile :
    logger . debug ( "Unpickling Model instance..." )
    model = pickle . load ( modelPickleFile )
    logger . debug ( "Finished unpickling Model instance" )
# Tell the model to load extra data , if any , that was too big for pickling
model . _deSerializeExtraData ( extraDataDir = Model . _getModelExtraDataDir ( savedModelDir ) )
logger . debug ( "Finished Loading model from local checkpoint" )
return model


signature:
def _getModelPickleFilePath ( saveModelDir ) :
    """Return the absolute path of the model ' s pickle file .
 : param saveModelDir : ( string )
 Directory of where the experiment is to be or was saved
 : returns : ( string ) An absolute path ."""

implementation:
path = os . path . join ( saveModelDir , "model.pkl" )
path = os . path . abspath ( path )
return path


signature:
def _getModelExtraDataDir ( saveModelDir ) :
    """Return the absolute path to the directory where the model ' s own
 " extra data " are stored ( i . e . , data that ' s too big for pickling ) .
 : param saveModelDir : ( string )
 Directory of where the experiment is to be or was saved
 : returns : ( string ) An absolute path ."""

implementation:
path = os . path . join ( saveModelDir , "modelextradata" )
path = os . path . abspath ( path )
return path


signature:
def runExperiment ( args , model = None ) :
    """Run a single OPF experiment .
 . . note : : The caller is responsible for initializing python logging before
 calling this function ( e . g . , import : mod : ` nupic . support ` ;
 : meth : ` nupic . support . initLogging ` )
 See also : : meth : ` . initExperimentPrng ` .
 : param args : ( string ) Experiment command - line args list . Too see all options ,
 run with ` ` - - help ` ` :
 . . code - block : : text
 Options :
 - h , - - help show this help message and exit
 - c < CHECKPOINT > Create a model and save it under the given < CHECKPOINT >
 name , but don ' t run it
 - - listCheckpoints List all available checkpoints
 - - listTasks List all task labels in description . py
 - - load = < CHECKPOINT > Load a model from the given < CHECKPOINT > and run it .
 Run with - - listCheckpoints flag for more details .
 - - newSerialization Use new capnproto serialization
 - - tasks Run the tasks with the given TASK LABELS in the order
 they are given . Either end of arg - list , or a
 standalone dot ( ' . ' ) arg or the next short or long
 option name ( - a or - - blah ) terminates the list . NOTE :
 FAILS TO RECOGNIZE task label names with one or more
 leading dashes . [ default : run all of the tasks in
 description . py ]
 - - testMode Reduce iteration count for testing
 - - noCheckpoint Don ' t checkpoint the model after running each task .
 : param model : ( : class : ` ~ nupic . frameworks . opf . model . Model ` ) For testing , may
 pass in an existing OPF Model to use instead of creating a new one .
 : returns : ( : class : ` ~ nupic . frameworks . opf . model . Model ` )
 reference to OPF Model instance that was constructed ( this
 is provided to aid with debugging ) or None , if none was
 created ."""

implementation:
# Parse command - line options
opt = _parseCommandLineOptions ( args )
# print " runExperiment : Parsed Command Options : " , opt
model = _runExperimentImpl ( opt , model )
return model


signature:
def _parseCommandLineOptions ( args ) :
    """Parse command line options
 Args :
 args : command line arguments ( not including sys . argv [ 0 ] )
 Returns :
 namedtuple ParseCommandLineOptionsResult"""

implementation:
usageStr = ( "%prog [options] descriptionPyDirectory\n" "This script runs a single OPF Model described by description.py " "located in the given directory." )
parser = optparse . OptionParser ( usage = usageStr )
parser . add_option ( "-c" , help = "Create a model and save it under the given " "<CHECKPOINT> name, but don't run it" , dest = "createCheckpointName" , action = "store" , type = "string" , default = "" , metavar = "<CHECKPOINT>" )
parser . add_option ( "--listCheckpoints" , help = "List all available checkpoints" , dest = "listAvailableCheckpoints" , action = "store_true" , default = False )
parser . add_option ( "--listTasks" , help = "List all task labels in description.py" , dest = "listTasks" , action = "store_true" , default = False )
parser . add_option ( "--load" , help = "Load a model from the given <CHECKPOINT> and run it. " "Run with --listCheckpoints flag for more details. " , dest = "runCheckpointName" , action = "store" , type = "string" , default = "" , metavar = "<CHECKPOINT>" )
parser . add_option ( "--newSerialization" , help = "Use new capnproto serialization" , dest = "newSerialization" , action = "store_true" , default = False )
# parser . add _ option ( " - - reuseDatasets " ,
# help = " Keep existing generated / aggregated datasets " ,
# dest = " reuseDatasets " , action = " store _ true " ,
# default = False )
parser . add_option ( "--tasks" , help = "Run the tasks with the given TASK LABELS " "in the order they are given.  Either end of " "arg-list, or a standalone dot ('.') arg or " "the next short or long option name (-a or " "--blah) terminates the list. NOTE: FAILS " "TO RECOGNIZE task label names with one or more " "leading dashes. [default: run all of the tasks in " "description.py]" , dest = "taskLabels" , default = [ ] , action = "callback" , callback = reapVarArgsCallback , metavar = "TASK_LABELS" )
parser . add_option ( "--testMode" , help = "Reduce iteration count for testing" , dest = "testMode" , action = "store_true" , default = False )
parser . add_option ( "--noCheckpoint" , help = "Don't checkpoint the model after running each task." , dest = "checkpointModel" , action = "store_false" , default = True )
options , experiments = parser . parse_args ( args )
# Validate args
mutuallyExclusiveOptionCount = sum ( [ bool ( options . createCheckpointName ) , options . listAvailableCheckpoints , options . listTasks , bool ( options . runCheckpointName ) ] )
if mutuallyExclusiveOptionCount > 1 :
    _reportCommandLineUsageErrorAndExit ( parser , "Options: -c, --listCheckpoints, --listTasks, and --load are " "mutually exclusive. Please select only one" )
mutuallyExclusiveOptionCount = sum ( [ bool ( not options . checkpointModel ) , bool ( options . createCheckpointName ) ] )
if mutuallyExclusiveOptionCount > 1 :
    _reportCommandLineUsageErrorAndExit ( parser , "Options: -c and --noCheckpoint are " "mutually exclusive. Please select only one" )
if len ( experiments ) != 1 :
    _reportCommandLineUsageErrorAndExit ( parser , "Exactly ONE experiment must be specified, but got %s (%s)" % ( len ( experiments ) , experiments ) )
# Done with parser
parser . destroy ( )
# Prepare results
# Directory path of the experiment ( that contain description . py )
experimentDir = os . path . abspath ( experiments [ 0 ] )
# RunExperiment . py ' s private options ( g _ parsedPrivateCommandLineOptionsSchema )
privateOptions = dict ( )
privateOptions [ 'createCheckpointName' ] = options . createCheckpointName
privateOptions [ 'listAvailableCheckpoints' ] = options . listAvailableCheckpoints
privateOptions [ 'listTasks' ] = options . listTasks
privateOptions [ 'runCheckpointName' ] = options . runCheckpointName
privateOptions [ 'newSerialization' ] = options . newSerialization
privateOptions [ 'testMode' ] = options . testMode
# privateOptions [ ' reuseDatasets ' ] = options . reuseDatasets
privateOptions [ 'taskLabels' ] = options . taskLabels
privateOptions [ 'checkpointModel' ] = options . checkpointModel
result = ParseCommandLineOptionsResult ( experimentDir = experimentDir , privateOptions = privateOptions )
return result


signature:
def reapVarArgsCallback ( option , optStr , value , parser ) :
    """Used as optparse callback for reaping a variable number of option args .
 The option may be specified multiple times , and all the args associated with
 that option name will be accumulated in the order that they are encountered"""

implementation:
newValues = [ ]
# Reap the args , taking care to stop before the next option or ' . '
gotDot = False
for arg in parser . rargs : # Stop on - - longname options
    if arg . startswith ( "--" ) and len ( arg ) > 2 :
        break
    # Stop on - b options
    if arg . startswith ( "-" ) and len ( arg ) > 1 :
        break
    if arg == "." :
        gotDot = True
        break
    newValues . append ( arg )
if not newValues :
    raise optparse . OptionValueError ( ( "Empty arg list for option %r expecting one or more args " "(remaining tokens: %r)" ) % ( optStr , parser . rargs ) )
del parser . rargs [ : len ( newValues ) + int ( gotDot ) ]
# Retrieve the existing arg accumulator , if any
value = getattr ( parser . values , option . dest , [ ] )
# print " Previous value : % r " % value
if value is None :
    value = [ ]
# Append the new args to the existing ones and save to the parser
value . extend ( newValues )
setattr ( parser . values , option . dest , value )


signature:
def _reportCommandLineUsageErrorAndExit ( parser , message ) :
    """Report usage error and exit program with error indication ."""

implementation:
print parser . get_usage ( )
print message
sys . exit ( 1 )


signature:
def _runExperimentImpl ( options , model = None ) :
    """Creates and runs the experiment
 Args :
 options : namedtuple ParseCommandLineOptionsResult
 model : For testing : may pass in an existing OPF Model instance
 to use instead of creating a new one .
 Returns : reference to OPFExperiment instance that was constructed ( this
 is provided to aid with debugging ) or None , if none was
 created ."""

implementation:
json_helpers . validate ( options . privateOptions , schemaDict = g_parsedPrivateCommandLineOptionsSchema )
# Load the experiment ' s description . py module
experimentDir = options . experimentDir
descriptionPyModule = helpers . loadExperimentDescriptionScriptFromDir ( experimentDir )
expIface = helpers . getExperimentDescriptionInterfaceFromModule ( descriptionPyModule )
# Handle " list checkpoints " request
if options . privateOptions [ 'listAvailableCheckpoints' ] :
    _printAvailableCheckpoints ( experimentDir )
    return None
# Load experiment tasks
experimentTasks = expIface . getModelControl ( ) . get ( 'tasks' , [ ] )
# If the tasks list is empty , and this is a nupic environment description
# file being run from the OPF , convert it to a simple OPF description file .
if ( len ( experimentTasks ) == 0 and expIface . getModelControl ( ) [ 'environment' ] == OpfEnvironment . Nupic ) :
    expIface . convertNupicEnvToOPF ( )
    experimentTasks = expIface . getModelControl ( ) . get ( 'tasks' , [ ] )
# Ensures all the source locations are either absolute paths or relative to
# the nupic . datafiles package _ data location .
expIface . normalizeStreamSources ( )
# Extract option
newSerialization = options . privateOptions [ 'newSerialization' ]
# Handle listTasks
if options . privateOptions [ 'listTasks' ] :
    print "Available tasks:"
    for label in [ t [ 'taskLabel' ] for t in experimentTasks ] :
        print "\t" , label
    return None
# Construct the experiment instance
if options . privateOptions [ 'runCheckpointName' ] :
    assert model is None
    checkpointName = options . privateOptions [ 'runCheckpointName' ]
    model = ModelFactory . loadFromCheckpoint ( savedModelDir = _getModelCheckpointDir ( experimentDir , checkpointName ) , newSerialization = newSerialization )
elif model is not None :
    print "Skipping creation of OPFExperiment instance: caller provided his own"
else :
    modelDescription = expIface . getModelDescription ( )
    model = ModelFactory . create ( modelDescription )
# Handle " create model " request
if options . privateOptions [ 'createCheckpointName' ] :
    checkpointName = options . privateOptions [ 'createCheckpointName' ]
    _saveModel ( model = model , experimentDir = experimentDir , checkpointLabel = checkpointName , newSerialization = newSerialization )
    return model
# Build the task list
# Default task execution index list is in the natural list order of the tasks
taskIndexList = range ( len ( experimentTasks ) )
customTaskExecutionLabelsList = options . privateOptions [ 'taskLabels' ]
if customTaskExecutionLabelsList :
    taskLabelsList = [ t [ 'taskLabel' ] for t in experimentTasks ]
    taskLabelsSet = set ( taskLabelsList )
    customTaskExecutionLabelsSet = set ( customTaskExecutionLabelsList )
    assert customTaskExecutionLabelsSet . issubset ( taskLabelsSet ) , ( "Some custom-provided task execution labels don't correspond " "to actual task labels: mismatched labels: %r; actual task " "labels: %r." ) % ( customTaskExecutionLabelsSet - taskLabelsSet , customTaskExecutionLabelsList )
    taskIndexList = [ taskLabelsList . index ( label ) for label in customTaskExecutionLabelsList ]
    print "#### Executing custom task list: %r" % [ taskLabelsList [ i ] for i in taskIndexList ]
# Run all experiment tasks
for taskIndex in taskIndexList :
    task = experimentTasks [ taskIndex ]
    # Create a task runner and run it !
    taskRunner = _TaskRunner ( model = model , task = task , cmdOptions = options )
    taskRunner . run ( )
    del taskRunner
    if options . privateOptions [ 'checkpointModel' ] :
        _saveModel ( model = model , experimentDir = experimentDir , checkpointLabel = task [ 'taskLabel' ] , newSerialization = newSerialization )
return model


signature:
def _saveModel ( model , experimentDir , checkpointLabel , newSerialization = False ) :
    """Save model"""

implementation:
checkpointDir = _getModelCheckpointDir ( experimentDir , checkpointLabel )
if newSerialization :
    model . writeToCheckpoint ( checkpointDir )
else :
    model . save ( saveModelDir = checkpointDir )


signature:
def _getModelCheckpointDir ( experimentDir , checkpointLabel ) :
    """Creates directory for serialization of the model
 checkpointLabel :
 Checkpoint label ( string )
 Returns :
 absolute path to the serialization directory"""

implementation:
checkpointDir = os . path . join ( getCheckpointParentDir ( experimentDir ) , checkpointLabel + g_defaultCheckpointExtension )
checkpointDir = os . path . abspath ( checkpointDir )
return checkpointDir


signature:
def getCheckpointParentDir ( experimentDir ) :
    """Get checkpoint parent dir .
 Returns : absolute path to the base serialization directory within which
 model checkpoints for this experiment are created"""

implementation:
baseDir = os . path . join ( experimentDir , "savedmodels" )
baseDir = os . path . abspath ( baseDir )
return baseDir


signature:
def _checkpointLabelFromCheckpointDir ( checkpointDir ) :
    """Returns a checkpoint label string for the given model checkpoint directory
 checkpointDir : relative or absolute model checkpoint directory path"""

implementation:
assert checkpointDir . endswith ( g_defaultCheckpointExtension )
lastSegment = os . path . split ( checkpointDir ) [ 1 ]
checkpointLabel = lastSegment [ 0 : - len ( g_defaultCheckpointExtension ) ]
return checkpointLabel


signature:
def _isCheckpointDir ( checkpointDir ) :
    """Return true iff checkpointDir appears to be a checkpoint directory ."""

implementation:
lastSegment = os . path . split ( checkpointDir ) [ 1 ]
if lastSegment [ 0 ] == '.' :
    return False
if not checkpointDir . endswith ( g_defaultCheckpointExtension ) :
    return False
if not os . path . isdir ( checkpointDir ) :
    return False
return True


signature:
def _printAvailableCheckpoints ( experimentDir ) :
    """List available checkpoints for the specified experiment ."""

implementation:
checkpointParentDir = getCheckpointParentDir ( experimentDir )
if not os . path . exists ( checkpointParentDir ) :
    print "No available checkpoints."
    return
checkpointDirs = [ x for x in os . listdir ( checkpointParentDir ) if _isCheckpointDir ( os . path . join ( checkpointParentDir , x ) ) ]
if not checkpointDirs :
    print "No available checkpoints."
    return
print "Available checkpoints:"
checkpointList = [ _checkpointLabelFromCheckpointDir ( x ) for x in checkpointDirs ]
for checkpoint in sorted ( checkpointList ) :
    print "\t" , checkpoint
print
print "To start from a checkpoint:"
print "  python run_opf_experiment.py experiment --load <CHECKPOINT>"
print "For example, to start from the checkpoint \"MyCheckpoint\":"
print "  python run_opf_experiment.py experiment --load MyCheckpoint"


signature:
def run ( self ) :
    """Runs a single experiment task"""

implementation:
self . __logger . debug ( "run(): Starting task <%s>" , self . __task [ 'taskLabel' ] )
# Set up the task
# Create our main loop - control iterator
if self . __cmdOptions . privateOptions [ 'testMode' ] :
    numIters = 10
else :
    numIters = self . __task [ 'iterationCount' ]
if numIters >= 0 :
    iterTracker = iter ( xrange ( numIters ) )
else :
    iterTracker = iter ( itertools . count ( ) )
# Initialize periodic activities
periodic = PeriodicActivityMgr ( requestedActivities = self . _createPeriodicActivities ( ) )
# Reset sequence states in the model , so it starts looking for a new
# sequence
# TODO : should this be done in OPFTaskDriver . setup ( ) , instead ? Is it always
# desired in Nupic ?
self . __model . resetSequenceStates ( )
# Have Task Driver perform its initial setup activities , including setup
# callbacks
self . __taskDriver . setup ( )
# Run it !
while True : # Check controlling iterator first
    try :
        next ( iterTracker )
    except StopIteration :
        break
    # Read next input record
    try :
        inputRecord = self . __datasetReader . next ( )
    except StopIteration :
        break
    # Process input record
    result = self . __taskDriver . handleInputRecord ( inputRecord = inputRecord )
    if InferenceElement . encodings in result . inferences :
        result . inferences . pop ( InferenceElement . encodings )
    self . __predictionLogger . writeRecord ( result )
    # Run periodic activities
    periodic . tick ( )
# Dump the experiment metrics at the end of the task
self . _getAndEmitExperimentMetrics ( final = True )
# Have Task Driver perform its final activities
self . __taskDriver . finalize ( )
# Reset sequence states in the model , so it starts looking for a new
# sequence
# TODO : should this be done in OPFTaskDriver . setup ( ) , instead ? Is it always
# desired in Nupic ?
self . __model . resetSequenceStates ( )


signature:
def _createPeriodicActivities ( self ) :
    """Creates and returns a list of activites for this TaskRunner instance
 Returns : a list of PeriodicActivityRequest elements"""

implementation:
# Initialize periodic activities
periodicActivities = [ ]
# Metrics reporting
class MetricsReportCb ( object ) :
    def __init__ ( self , taskRunner ) :
        self . __taskRunner = taskRunner
        return
    def __call__ ( self ) :
        self . __taskRunner . _getAndEmitExperimentMetrics ( )
reportMetrics = PeriodicActivityRequest ( repeating = True , period = 1000 , cb = MetricsReportCb ( self ) )
periodicActivities . append ( reportMetrics )
# Iteration progress
class IterationProgressCb ( object ) :
    PROGRESS_UPDATE_PERIOD_TICKS = 1000
    def __init__ ( self , taskLabel , requestedIterationCount , logger ) :
        self . __taskLabel = taskLabel
        self . __requestedIterationCount = requestedIterationCount
        self . __logger = logger
        self . __numIterationsSoFar = 0
    def __call__ ( self ) :
        self . __numIterationsSoFar += self . PROGRESS_UPDATE_PERIOD_TICKS
        self . __logger . debug ( "%s: ITERATION PROGRESS: %s of %s" % ( self . __taskLabel , self . __numIterationsSoFar , self . __requestedIterationCount ) )
iterationProgressCb = IterationProgressCb ( taskLabel = self . __task [ 'taskLabel' ] , requestedIterationCount = self . __task [ 'iterationCount' ] , logger = self . __logger )
iterationProgressReporter = PeriodicActivityRequest ( repeating = True , period = IterationProgressCb . PROGRESS_UPDATE_PERIOD_TICKS , cb = iterationProgressCb )
periodicActivities . append ( iterationProgressReporter )
return periodicActivities


signature:
def _generateFile ( filename , data ) :
    """Parameters :
 filename : name of . csv file to generate"""

implementation:
# Create the file
print "Creating %s..." % ( filename )
numRecords , numFields = data . shape
fields = [ ( 'field%d' % ( i + 1 ) , 'float' , '' ) for i in range ( numFields ) ]
outFile = File ( filename , fields )
for i in xrange ( numRecords ) :
    outFile . write ( data [ i ] . tolist ( ) )
outFile . close ( )