signature:
def index_collection ( self , filenames ) :
    "Index a whole collection of files ."

implementation:
for filename in filenames :
    self . index_document ( open ( filename ) . read ( ) , filename )


signature:
def index_document ( self , text , url ) :
    "Index the text of a document ."

implementation:
# # For now , use first line for title
title = text [ : text . index ( '\n' ) ] . strip ( )
docwords = words ( text )
docid = len ( self . documents )
self . documents . append ( Document ( title , url , len ( docwords ) ) )
for word in docwords :
    if word not in self . stopwords :
        self . index [ word ] [ docid ] += 1


signature:
def query ( self , query_text , n = 10 ) :
    """Return a list of n ( score , docid ) pairs for the best matches .
 Also handle the special syntax for ' learn : command ' ."""

implementation:
if query_text . startswith ( "learn:" ) :
    doctext = os . popen ( query_text [ len ( "learn:" ) : ] , 'r' ) . read ( )
    self . index_document ( doctext , query_text )
    return [ ]
qwords = [ w for w in words ( query_text ) if w not in self . stopwords ]
shortest = argmin ( qwords , lambda w : len ( self . index [ w ] ) )
docs = self . index [ shortest ]
results = [ ( sum ( [ self . score ( w , d ) for w in qwords ] ) , d ) for d in docs ]
results . sort ( ) ;
results . reverse ( )
return results [ : n ]


signature:
def score ( self , word , docid ) :
    "Compute a score for this word on this docid ."

implementation:
# # There are many options ; here we take a very simple approach
return ( math . log ( 1 + self . index [ word ] [ docid ] ) / math . log ( 1 + self . documents [ docid ] . nwords ) )


signature:
def present ( self , results ) :
    "Present the results as a list ."

implementation:
for ( score , d ) in results :
    doc = self . documents [ d ]
    print ( "%5.2f|%25s | %s" % ( 100 * score , doc . url , doc . title [ : 45 ] . expandtabs ( ) ) )


signature:
def present_results ( self , query_text , n = 10 ) :
    "Get results for the query and present them ."

implementation: self . present ( self . query ( query_text , n ) )


signature:
def score ( self , plaintext ) :
    "Return a score for text based on how common letters pairs are ."

implementation:
s = 1.0
for bi in bigrams ( plaintext ) :
    s = s * self . P2 [ bi ]
return s


signature:
def decode ( self , ciphertext ) :
    "Search for a decoding of the ciphertext ."

implementation:
self . ciphertext = ciphertext
problem = PermutationDecoderProblem ( decoder = self )
return search . best_first_tree_search ( problem , lambda node : self . score ( node . state ) )


signature:
def score ( self , code ) :
    """Score is product of word scores , unigram scores , and bigram scores .
 This can get very small , so we use logs and exp ."""

implementation:
text = permutation_decode ( self . ciphertext , code )
logP = ( sum ( [ log ( self . Pwords [ word ] ) for word in words ( text ) ] ) + sum ( [ log ( self . P1 [ c ] ) for c in text ] ) + sum ( [ log ( self . P2 [ b ] ) for b in bigrams ( text ) ] ) )
return exp ( logP )


signature:
def get_value ( self , context , default ) :
    """Returns a ` ` SettingDict ` ` object ."""

implementation:
if default is None :
    settings = self . setting_model . objects . as_dict ( )
else :
    settings = self . setting_model . objects . as_dict ( default = default )
return settings


signature:
def get_value ( self , context , name , default ) :
    """Returns the value of the named setting ."""

implementation:
settings = self . setting_model . objects . filter ( name = name )
if default is None :
    settings = settings . as_dict ( )
else :
    settings = settings . as_dict ( default = default )
value = settings [ name ]
return value


signature:
def render_tag ( self , context , name , nodelist ) :
    """Returns the value of the named setting ."""

implementation:
# Use ` try ` and ` except ` instead of ` setdefault ( ) ` so we can skip
# rendering the nodelist when the setting already exists .
settings = self . setting_model . objects . filter ( name = name ) . as_dict ( )
try :
    value = settings [ name ]
except KeyError :
    value = settings [ name ] = nodelist . render ( context )
return value


signature:
def value_iteration ( mdp , epsilon = 0.001 ) :
    "Solving an MDP by value iteration . [ Fig . 17.4]"

implementation:
U1 = dict ( [ ( s , 0 ) for s in mdp . states ] )
R , T , gamma = mdp . R , mdp . T , mdp . gamma
while True :
    U = U1 . copy ( )
    delta = 0
    for s in mdp . states :
        U1 [ s ] = R ( s ) + gamma * max ( [ sum ( [ p * U [ s1 ] for ( p , s1 ) in T ( s , a ) ] ) for a in mdp . actions ( s ) ] )
        delta = max ( delta , abs ( U1 [ s ] - U [ s ] ) )
    if delta < epsilon * ( 1 - gamma ) / gamma :
        return U


signature:
def best_policy ( mdp , U ) :
    """Given an MDP and a utility function U , determine the best policy ,
 as a mapping from state to action . ( Equation 17.4)"""

implementation:
pi = { }
for s in mdp . states :
    pi [ s ] = argmax ( mdp . actions ( s ) , lambda a : expected_utility ( a , s , U , mdp ) )
return pi


signature:
def expected_utility ( a , s , U , mdp ) :
    "The expected utility of doing a in state s , according to the MDP and U ."

implementation: return sum ( [ p * U [ s1 ] for ( p , s1 ) in mdp . T ( s , a ) ] )


signature:
def policy_iteration ( mdp ) :
    "Solve an MDP by policy iteration [ Fig . 17.7]"

implementation:
U = dict ( [ ( s , 0 ) for s in mdp . states ] )
pi = dict ( [ ( s , random . choice ( mdp . actions ( s ) ) ) for s in mdp . states ] )
while True :
    U = policy_evaluation ( pi , U , mdp )
    unchanged = True
    for s in mdp . states :
        a = argmax ( mdp . actions ( s ) , lambda a : expected_utility ( a , s , U , mdp ) )
        if a != pi [ s ] :
            pi [ s ] = a
            unchanged = False
    if unchanged :
        return pi


signature:
def policy_evaluation ( pi , U , mdp , k = 20 ) :
    """Return an updated utility mapping U from each state in the MDP to its
 utility , using an approximation ( modified policy iteration ) ."""

implementation:
R , T , gamma = mdp . R , mdp . T , mdp . gamma
for i in range ( k ) :
    for s in mdp . states :
        U [ s ] = R ( s ) + gamma * sum ( [ p * U [ s1 ] for ( p , s1 ) in T ( s , pi [ s ] ) ] )
return U


signature:
def go ( self , state , direction ) :
    "Return the state that results from going in this direction ."

implementation:
state1 = vector_add ( state , direction )
return if_ ( state1 in self . states , state1 , state )


signature:
def to_grid ( self , mapping ) :
    """Convert a mapping from ( x , y ) to v into a [ [ . . . , v , . . . ] ] grid ."""

implementation: return list ( reversed ( [ [ mapping . get ( ( x , y ) , None ) for x in range ( self . cols ) ] for y in range ( self . rows ) ] ) )


signature:
def as_dict ( self , default = None ) :
    """Returns a ` ` SettingDict ` ` object for this queryset ."""

implementation:
settings = SettingDict ( queryset = self , default = default )
return settings


signature:
def create ( self , name , value ) :
    """Creates and returns an object of the appropriate type for ` ` value ` ` ."""

implementation:
if value is None :
    raise ValueError ( 'Setting value cannot be `None`.' )
model = Setting . get_model_for_value ( value )
# Call ` create ( ) ` method on the super class to avoid recursion .
obj = super ( SettingQuerySet , model . objects . all ( ) ) . create ( name = name , value = value )
return obj


signature:
def get_model_for_value ( cls , value ) :
    """Iterates through setting value subclasses , returning one that is
 compatible with the type of ` ` value ` ` . Calls ` ` is _ compatible ( ) ` ` on
 each subclass ."""

implementation:
for related_object in get_all_related_objects ( cls . _meta ) :
    model = getattr ( related_object , 'related_model' , related_object . model )
    if issubclass ( model , cls ) :
        if model . is_compatible ( value ) :
            return model
raise ValueError ( 'No compatible `SettingValueModel` subclass for %r' % value )


signature:
def is_compatible ( cls , value ) :
    """Returns ` ` True ` ` if this model should be used to store ` ` value ` ` .
 Checks if ` ` value ` ` is an instance of ` ` value _ type ` ` . Override this
 method if you need more advanced behaviour . For example , to distinguish
 between single and multi - line text ."""

implementation:
if not hasattr ( cls , 'value_type' ) :
    raise NotImplementedError ( 'You must define a `value_type` attribute or override the ' '`is_compatible()` method on `SettingValueModel` subclasses.' )
return isinstance ( value , cls . value_type )


signature:
def tree_search ( problem , frontier ) :
    """Search through the successors of a problem to find a goal .
 The argument frontier should be an empty queue .
 Don ' t worry about repeated paths to a state . [ Fig . 3.7]"""

implementation:
frontier . append ( Node ( problem . initial ) )
while frontier :
    node = frontier . pop ( )
    if problem . goal_test ( node . state ) :
        return node
    frontier . extend ( node . expand ( problem ) )
return None


signature:
def graph_search ( problem , frontier ) :
    """Search through the successors of a problem to find a goal .
 The argument frontier should be an empty queue .
 If two paths reach a state , only use the first one . [ Fig . 3.7]"""

implementation:
frontier . append ( Node ( problem . initial ) )
explored = set ( )
while frontier :
    node = frontier . pop ( )
    if problem . goal_test ( node . state ) :
        return node
    explored . add ( node . state )
    frontier . extend ( child for child in node . expand ( problem ) if child . state not in explored and child not in frontier )
return None


signature:
def breadth_first_search ( problem ) :
    "[ Fig . 3.11]"

implementation:
node = Node ( problem . initial )
if problem . goal_test ( node . state ) :
    return node
frontier = FIFOQueue ( )
frontier . append ( node )
explored = set ( )
while frontier :
    node = frontier . pop ( )
    explored . add ( node . state )
    for child in node . expand ( problem ) :
        if child . state not in explored and child not in frontier :
            if problem . goal_test ( child . state ) :
                return child
            frontier . append ( child )
return None


signature:
def best_first_graph_search ( problem , f ) :
    """Search the nodes with the lowest f scores first .
 You specify the function f ( node ) that you want to minimize ; for example ,
 if f is a heuristic estimate to the goal , then we have greedy best
 first search ; if f is node . depth then we have breadth - first search .
 There is a subtlety : the line " f = memoize ( f , ' f ' ) " means that the f
 values will be cached on the nodes as they are computed . So after doing
 a best first search you can examine the f values of the path returned ."""

implementation:
f = memoize ( f , 'f' )
node = Node ( problem . initial )
if problem . goal_test ( node . state ) :
    return node
frontier = PriorityQueue ( min , f )
frontier . append ( node )
explored = set ( )
while frontier :
    node = frontier . pop ( )
    if problem . goal_test ( node . state ) :
        return node
    explored . add ( node . state )
    for child in node . expand ( problem ) :
        if child . state not in explored and child not in frontier :
            frontier . append ( child )
        elif child in frontier :
            incumbent = frontier [ child ]
            if f ( child ) < f ( incumbent ) :
                del frontier [ incumbent ]
                frontier . append ( child )
return None


signature:
def depth_limited_search ( problem , limit = 50 ) :
    "[ Fig . 3.17]"

implementation:
def recursive_dls ( node , problem , limit ) :
    if problem . goal_test ( node . state ) :
        return node
    elif node . depth == limit :
        return 'cutoff'
    else :
        cutoff_occurred = False
        for child in node . expand ( problem ) :
            result = recursive_dls ( child , problem , limit )
            if result == 'cutoff' :
                cutoff_occurred = True
            elif result is not None :
                return result
        return if_ ( cutoff_occurred , 'cutoff' , None )
# Body of depth _ limited _ search :
return recursive_dls ( Node ( problem . initial ) , problem , limit )


signature:
def iterative_deepening_search ( problem ) :
    "[ Fig . 3.18]"

implementation:
for depth in xrange ( sys . maxint ) :
    result = depth_limited_search ( problem , depth )
    if result != 'cutoff' :
        return result


signature:
def astar_search ( problem , h = None ) :
    """A * search is best - first graph search with f ( n ) = g ( n ) + h ( n ) .
 You need to specify the h function when you call astar _ search , or
 else in your Problem subclass ."""

implementation:
h = memoize ( h or problem . h , 'h' )
return best_first_graph_search ( problem , lambda n : n . path_cost + h ( n ) )


signature:
def recursive_best_first_search ( problem , h = None ) :
    "[ Fig . 3.26]"

implementation:
h = memoize ( h or problem . h , 'h' )
def RBFS ( problem , node , flimit ) :
    if problem . goal_test ( node . state ) :
        return node , 0
        # ( The second value is immaterial )
    successors = node . expand ( problem )
    if len ( successors ) == 0 :
        return None , infinity
    for s in successors :
        s . f = max ( s . path_cost + h ( s ) , node . f )
    while True :
        successors . sort ( lambda x , y : cmp ( x . f , y . f ) )
        # Order by lowest f value
        best = successors [ 0 ]
        if best . f > flimit :
            return None , best . f
        if len ( successors ) > 1 :
            alternative = successors [ 1 ] . f
        else :
            alternative = infinity
        result , best . f = RBFS ( problem , best , min ( flimit , alternative ) )
        if result is not None :
            return result , best . f
node = Node ( problem . initial )
node . f = h ( node )
result , bestf = RBFS ( problem , node , infinity )
return result


signature:
def hill_climbing ( problem ) :
    """From the initial node , keep choosing the neighbor with highest value ,
 stopping when no neighbor is better . [ Fig . 4.2]"""

implementation:
current = Node ( problem . initial )
while True :
    neighbors = current . expand ( problem )
    if not neighbors :
        break
    neighbor = argmax_random_tie ( neighbors , lambda node : problem . value ( node . state ) )
    if problem . value ( neighbor . state ) <= problem . value ( current . state ) :
        break
    current = neighbor
return current . state


signature:
def exp_schedule ( k = 20 , lam = 0.005 , limit = 100 ) :
    "One possible schedule function for simulated annealing"

implementation: return lambda t : if_ ( t < limit , k * math . exp ( - lam * t ) , 0 )


signature:
def simulated_annealing ( problem , schedule = exp_schedule ( ) ) :
    "[ Fig . 4.5]"

implementation:
current = Node ( problem . initial )
for t in xrange ( sys . maxint ) :
    T = schedule ( t )
    if T == 0 :
        return current
    neighbors = current . expand ( problem )
    if not neighbors :
        return current
    next = random . choice ( neighbors )
    delta_e = problem . value ( next . state ) - problem . value ( current . state )
    if delta_e > 0 or probability ( math . exp ( delta_e / T ) ) :
        current = next


signature:
def genetic_search ( problem , fitness_fn , ngen = 1000 , pmut = 0.1 , n = 20 ) :
    """Call genetic _ algorithm on the appropriate parts of a problem .
 This requires the problem to have states that can mate and mutate ,
 plus a value method that scores states ."""

implementation:
s = problem . initial_state
states = [ problem . result ( s , a ) for a in problem . actions ( s ) ]
random . shuffle ( states )
return genetic_algorithm ( states [ : n ] , problem . value , ngen , pmut )


signature:
def genetic_algorithm ( population , fitness_fn , ngen = 1000 , pmut = 0.1 ) :
    "[ Fig . 4.8]"

implementation:
for i in range ( ngen ) :
    new_population = [ ]
    for i in len ( population ) :
        fitnesses = map ( fitness_fn , population )
        p1 , p2 = weighted_sample_with_replacement ( population , fitnesses , 2 )
        child = p1 . mate ( p2 )
        if random . uniform ( 0 , 1 ) < pmut :
            child . mutate ( )
        new_population . append ( child )
    population = new_population
return argmax ( population , fitness_fn )


signature:
def RandomGraph ( nodes = range ( 10 ) , min_links = 2 , width = 400 , height = 300 , curvature = lambda : random . uniform ( 1.1 , 1.5 ) ) :
    """Construct a random graph , with the specified nodes , and random links .
 The nodes are laid out randomly on a ( width x height ) rectangle .
 Then each node is connected to the min _ links nearest neighbors .
 Because inverse links are added , some nodes will have more connections .
 The distance between nodes is the hypotenuse times curvature ( ) ,
 where curvature ( ) defaults to a random number between 1.1 and 1.5."""

implementation:
g = UndirectedGraph ( )
g . locations = { }
# # Build the cities
for node in nodes :
    g . locations [ node ] = ( random . randrange ( width ) , random . randrange ( height ) )
# # Build roads from each city to at least min _ links nearest neighbors .
for i in range ( min_links ) :
    for node in nodes :
        if len ( g . get ( node ) ) < min_links :
            here = g . locations [ node ]
            def distance_to_node ( n ) :
                if n is node or g . get ( node , n ) :
                    return infinity
                return distance ( g . locations [ n ] , here )
            neighbor = argmin ( nodes , distance_to_node )
            d = distance ( g . locations [ neighbor ] , here ) * curvature ( )
            g . connect ( node , neighbor , int ( d ) )
return g


signature:
def random_boggle ( n = 4 ) :
    """Return a random Boggle board of size n x n .
 We represent a board as a linear list of letters ."""

implementation:
cubes = [ cubes16 [ i % 16 ] for i in range ( n * n ) ]
random . shuffle ( cubes )
return map ( random . choice , cubes )


signature:
def print_boggle ( board ) :
    "Print the board in a 2 - d array ."

implementation:
n2 = len ( board ) ;
n = exact_sqrt ( n2 )
for i in range ( n2 ) :
    if i % n == 0 and i > 0 :
        print
    if board [ i ] == 'Q' :
        print 'Qu' ,
    else :
        print str ( board [ i ] ) + ' ' ,
print


signature:
def boggle_neighbors ( n2 , cache = { } ) :
    """Return a list of lists , where the i - th element is the list of indexes
 for the neighbors of square i ."""

implementation:
if cache . get ( n2 ) :
    return cache . get ( n2 )
n = exact_sqrt ( n2 )
neighbors = [ None ] * n2
for i in range ( n2 ) :
    neighbors [ i ] = [ ]
    on_top = i < n
    on_bottom = i >= n2 - n
    on_left = i % n == 0
    on_right = ( i + 1 ) % n == 0
    if not on_top :
        neighbors [ i ] . append ( i - n )
        if not on_left :
            neighbors [ i ] . append ( i - n - 1 )
        if not on_right :
            neighbors [ i ] . append ( i - n + 1 )
    if not on_bottom :
        neighbors [ i ] . append ( i + n )
        if not on_left :
            neighbors [ i ] . append ( i + n - 1 )
        if not on_right :
            neighbors [ i ] . append ( i + n + 1 )
    if not on_left :
        neighbors [ i ] . append ( i - 1 )
    if not on_right :
        neighbors [ i ] . append ( i + 1 )
cache [ n2 ] = neighbors
return neighbors


signature:
def exact_sqrt ( n2 ) :
    "If n2 is a perfect square , return its square root , else raise error ."

implementation:
n = int ( math . sqrt ( n2 ) )
assert n * n == n2
return n


signature:
def boggle_hill_climbing ( board = None , ntimes = 100 , verbose = True ) :
    """Solve inverse Boggle by hill - climbing : find a high - scoring board by
 starting with a random one and changing it ."""

implementation:
finder = BoggleFinder ( )
if board is None :
    board = random_boggle ( )
best = len ( finder . set_board ( board ) )
for _ in range ( ntimes ) :
    i , oldc = mutate_boggle ( board )
    new = len ( finder . set_board ( board ) )
    if new > best :
        best = new
        if verbose :
            print best , _ , board
    else :
        board [ i ] = oldc
        # # Change back
if verbose :
    print_boggle ( board )
return board , best


signature:
def compare_graph_searchers ( ) :
    """Prints a table of results like this :
 > > > compare _ graph _ searchers ( )
 Searcher Romania ( A , B ) Romania ( O , N ) Australia
 breadth _ first _ tree _ search < 21 / 22 / 59 / B > < 1158/1159/3288 / N > < 7 / 8 / 22 / WA >
 breadth _ first _ search < 7 / 11 / 18 / B > < 19 / 20 / 45 / N > < 2 / 6 / 8 / WA >
 depth _ first _ graph _ search < 8 / 9 / 20 / B > < 16 / 17 / 38 / N > < 4 / 5 / 11 / WA >
 iterative _ deepening _ search < 11 / 33 / 31 / B > < 656/1815/1812 / N > < 3 / 11 / 11 / WA >
 depth _ limited _ search < 54 / 65 / 185 / B > < 387/1012/1125 / N > < 50 / 54 / 200 / WA >
 recursive _ best _ first _ search < 5 / 6 / 15 / B > < 5887/5888/16532 / N > < 11 / 12 / 43 / WA >"""

implementation: compare_searchers ( problems = [ GraphProblem ( 'A' , 'B' , romania ) , GraphProblem ( 'O' , 'N' , romania ) , GraphProblem ( 'Q' , 'WA' , australia ) ] , header = [ 'Searcher' , 'Romania(A, B)' , 'Romania(O, N)' , 'Australia' ] )


signature:
def expand ( self , problem ) :
    "List the nodes reachable in one step from this node ."

implementation: return [ self . child_node ( problem , action ) for action in problem . actions ( self . state ) ]


signature:
def child_node ( self , problem , action ) :
    "Fig . 3.10"

implementation:
next = problem . result ( self . state , action )
return Node ( next , self , action , problem . path_cost ( self . path_cost , self . state , action , next ) )


signature:
def path ( self ) :
    "Return a list of nodes forming the path from the root to this node ."

implementation:
node , path_back = self , [ ]
while node :
    path_back . append ( node )
    node = node . parent
return list ( reversed ( path_back ) )


signature:
def mate ( self , other ) :
    "Return a new individual crossing self and other ."

implementation:
c = random . randrange ( len ( self . genes ) )
return self . __class__ ( self . genes [ : c ] + other . genes [ c : ] )


signature:
def make_undirected ( self ) :
    "Make a digraph into an undirected graph by adding symmetric edges ."

implementation:
for a in self . dict . keys ( ) :
    for ( b , distance ) in self . dict [ a ] . items ( ) :
        self . connect1 ( b , a , distance )


signature:
def connect ( self , A , B , distance = 1 ) :
    """Add a link from A and B of given distance , and also add the inverse
 link if the graph is undirected ."""

implementation:
self . connect1 ( A , B , distance )
if not self . directed :
    self . connect1 ( B , A , distance )


signature:
def connect1 ( self , A , B , distance ) :
    "Add a link from A to B of given distance , in one direction only ."

implementation: self . dict . setdefault ( A , { } ) [ B ] = distance


signature:
def get ( self , a , b = None ) :
    """Return a link distance or a dict of { node : distance } entries .
 . get ( a , b ) returns the distance or None ;
 . get ( a ) returns a dict of { node : distance } entries , possibly { } ."""

implementation:
links = self . dict . setdefault ( a , { } )
if b is None :
    return links
else :
    return links . get ( b )


signature:
def h ( self , node ) :
    "h function is straight - line distance from a node ' s state to goal ."

implementation:
locs = getattr ( self . graph , 'locations' , None )
if locs :
    return int ( distance ( locs [ node . state ] , locs [ self . goal ] ) )
else :
    return infinity


signature:
def actions ( self , state ) :
    "In the leftmost empty column , try all non - conflicting rows ."

implementation:
if state [ - 1 ] is not None :
    return [ ]
    # All columns filled ; no successors
else :
    col = state . index ( None )
    return [ row for row in range ( self . N ) if not self . conflicted ( state , row , col ) ]


signature:
def result ( self , state , row ) :
    "Place the next queen at the given row ."

implementation:
col = state . index ( None )
new = state [ : ]
new [ col ] = row
return new


signature:
def conflicted ( self , state , row , col ) :
    "Would placing a queen at ( row , col ) conflict with anything ?"

implementation: return any ( self . conflict ( row , col , state [ c ] , c ) for c in range ( col ) )


signature:
def conflict ( self , row1 , col1 , row2 , col2 ) :
    "Would putting two queens in ( row1 , col1 ) and ( row2 , col2 ) conflict ?"

implementation:
return ( row1 == row2 # # same row
or col1 == col2 # # same column
or row1 - col1 == row2 - col2 # # same \ diagonal
or row1 + col1 == row2 + col2 )


signature:
def lookup ( self , prefix , lo = 0 , hi = None ) :
    """See if prefix is in dictionary , as a full word or as a prefix .
 Return two values : the first is the lowest i such that
 words [ i ] . startswith ( prefix ) , or is None ; the second is
 True iff prefix itself is in the Wordlist ."""

implementation:
words = self . words
if hi is None :
    hi = len ( words )
i = bisect . bisect_left ( words , prefix , lo , hi )
if i < len ( words ) and words [ i ] . startswith ( prefix ) :
    return i , ( words [ i ] == prefix )
else :
    return None , False


signature:
def set_board ( self , board = None ) :
    "Set the board , and find all the words in it ."

implementation:
if board is None :
    board = random_boggle ( )
self . board = board
self . neighbors = boggle_neighbors ( len ( board ) )
self . found = { }
for i in range ( len ( board ) ) :
    lo , hi = self . wordlist . bounds [ board [ i ] ]
    self . find ( lo , hi , i , [ ] , '' )
return self


signature:
def find ( self , lo , hi , i , visited , prefix ) :
    """Looking in square i , find the words that continue the prefix ,
 considering the entries in self . wordlist . words [ lo : hi ] , and not
 revisiting the squares in visited ."""

implementation:
if i in visited :
    return
wordpos , is_word = self . wordlist . lookup ( prefix , lo , hi )
if wordpos is not None :
    if is_word :
        self . found [ prefix ] = True
    visited . append ( i )
    c = self . board [ i ]
    if c == 'Q' :
        c = 'QU'
    prefix += c
    for j in self . neighbors [ i ] :
        self . find ( wordpos , hi , j , visited , prefix )
    visited . pop ( )


signature:
def score ( self ) :
    "The total score for the words found , according to the rules ."

implementation: return sum ( [ self . scores [ len ( w ) ] for w in self . words ( ) ] )


signature:
def TraceAgent ( agent ) :
    """Wrap the agent ' s program to print its input and output . This will let
 you see what the agent is doing in the environment ."""

implementation:
old_program = agent . program
def new_program ( percept ) :
    action = old_program ( percept )
    print '%s perceives %s and does %s' % ( agent , percept , action )
    return action
agent . program = new_program
return agent


signature:
def TableDrivenAgentProgram ( table ) :
    """This agent selects an action based on the percept sequence .
 It is practical only for tiny domains .
 To customize it , provide as table a dictionary of all
 { percept _ sequence : action } pairs . [ Fig . 2.7]"""

implementation:
percepts = [ ]
def program ( percept ) :
    percepts . append ( percept )
    action = table . get ( tuple ( percepts ) )
    return action
return program


signature:
def SimpleReflexAgentProgram ( rules , interpret_input ) :
    "This agent takes action based solely on the percept . [ Fig . 2.10]"

implementation:
def program ( percept ) :
    state = interpret_input ( percept )
    rule = rule_match ( state , rules )
    action = rule . action
    return action
return program


signature:
def ModelBasedReflexAgentProgram ( rules , update_state ) :
    "This agent takes action based on the percept and state . [ Fig . 2.12]"

implementation:
def program ( percept ) :
    program . state = update_state ( program . state , program . action , percept )
    rule = rule_match ( program . state , rules )
    action = rule . action
    return action
program . state = program . action = None
return program


signature:
def TableDrivenVacuumAgent ( ) :
    "[ Fig . 2.3]"

implementation:
table = { ( ( loc_A , 'Clean' ) , ) : 'Right' , ( ( loc_A , 'Dirty' ) , ) : 'Suck' , ( ( loc_B , 'Clean' ) , ) : 'Left' , ( ( loc_B , 'Dirty' ) , ) : 'Suck' , ( ( loc_A , 'Clean' ) , ( loc_A , 'Clean' ) ) : 'Right' , ( ( loc_A , 'Clean' ) , ( loc_A , 'Dirty' ) ) : 'Suck' , ( ( loc_A , 'Clean' ) , ( loc_A , 'Clean' ) , ( loc_A , 'Clean' ) ) : 'Right' , ( ( loc_A , 'Clean' ) , ( loc_A , 'Clean' ) , ( loc_A , 'Dirty' ) ) : 'Suck' , }
return Agent ( TableDrivenAgentProgram ( table ) )


signature:
def ReflexVacuumAgent ( ) :
    "A reflex agent for the two - state vacuum environment . [ Fig . 2.8]"

implementation:
def program ( ( location , status ) ) :
    if status == 'Dirty' :
        return 'Suck'
    elif location == loc_A :
        return 'Right'
    elif location == loc_B :
        return 'Left'
return Agent ( program )


signature:
def compare_agents ( EnvFactory , AgentFactories , n = 10 , steps = 1000 ) :
    """See how well each of several agents do in n instances of an environment .
 Pass in a factory ( constructor ) for environments , and several for agents .
 Create n instances of the environment , and run each agent in copies of
 each one for steps . Return a list of ( agent , average - score ) tuples ."""

implementation:
envs = [ EnvFactory ( ) for i in range ( n ) ]
return [ ( A , test_agent ( A , steps , copy . deepcopy ( envs ) ) ) for A in AgentFactories ]


signature:
def step ( self ) :
    """Run the environment for one time step . If the
 actions and exogenous changes are independent , this method will
 do . If there are interactions between them , you ' ll need to
 override this method ."""

implementation:
if not self . is_done ( ) :
    actions = [ agent . program ( self . percept ( agent ) ) for agent in self . agents ]
    for ( agent , action ) in zip ( self . agents , actions ) :
        self . execute_action ( agent , action )
    self . exogenous_change ( )


signature:
def run ( self , steps = 1000 ) :
    "Run the Environment for given number of time steps ."

implementation:
for step in range ( steps ) :
    if self . is_done ( ) :
        return
    self . step ( )


signature:
def list_things_at ( self , location , tclass = Thing ) :
    "Return all things exactly at a given location ."

implementation: return [ thing for thing in self . things if thing . location == location and isinstance ( thing , tclass ) ]


signature:
def add_thing ( self , thing , location = None ) :
    """Add a thing to the environment , setting its location . For
 convenience , if thing is an agent program we make a new agent
 for it . ( Shouldn ' t need to override this ."""

implementation:
if not isinstance ( thing , Thing ) :
    thing = Agent ( thing )
assert thing not in self . things , "Don't add the same thing twice"
thing . location = location or self . default_location ( thing )
self . things . append ( thing )
if isinstance ( thing , Agent ) :
    thing . performance = 0
    self . agents . append ( thing )


signature:
def delete_thing ( self , thing ) :
    """Remove a thing from the environment ."""

implementation:
try :
    self . things . remove ( thing )
except ValueError , e :
    print e
    print "  in Environment delete_thing"
    print "  Thing to be removed: %s at %s" % ( thing , thing . location )
    print "  from list: %s" % [ ( thing , thing . location ) for thing in self . things ]
if thing in self . agents :
    self . agents . remove ( thing )


signature:
def things_near ( self , location , radius = None ) :
    "Return all things within radius of location ."

implementation:
if radius is None :
    radius = self . perceptible_distance
radius2 = radius * radius
return [ thing for thing in self . things if distance2 ( location , thing . location ) <= radius2 ]


signature:
def percept ( self , agent ) :
    "By default , agent perceives things within a default radius ."

implementation: return [ self . thing_percept ( thing , agent ) for thing in self . things_near ( agent . location ) ]


signature:
def move_to ( self , thing , destination ) :
    "Move a thing to a new location ."

implementation:
thing . bump = self . some_things_at ( destination , Obstacle )
if not thing . bump :
    thing . location = destination
    for o in self . observers :
        o . thing_moved ( thing )


signature:
def add_walls ( self ) :
    "Put walls around the entire perimeter of the grid ."

implementation:
for x in range ( self . width ) :
    self . add_thing ( Wall ( ) , ( x , 0 ) )
    self . add_thing ( Wall ( ) , ( x , self . height - 1 ) )
for y in range ( self . height ) :
    self . add_thing ( Wall ( ) , ( 0 , y ) )
    self . add_thing ( Wall ( ) , ( self . width - 1 , y ) )


signature:
def percept ( self , agent ) :
    """The percept is a tuple of ( ' Dirty ' or ' Clean ' , ' Bump ' or ' None ' ) .
 Unlike the TrivialVacuumEnvironment , location is NOT perceived ."""

implementation:
status = if_ ( self . some_things_at ( agent . location , Dirt ) , 'Dirty' , 'Clean' )
bump = if_ ( agent . bump , 'Bump' , 'None' )
return ( status , bump )


signature:
def execute_action ( self , agent , action ) :
    """Change agent ' s location and / or location ' s status ; track performance .
 Score 10 for each dirt cleaned ; - 1 for each move ."""

implementation:
if action == 'Right' :
    agent . location = loc_B
    agent . performance -= 1
elif action == 'Left' :
    agent . location = loc_A
    agent . performance -= 1
elif action == 'Suck' :
    if self . status [ agent . location ] == 'Dirty' :
        agent . performance += 10
    self . status [ agent . location ] = 'Clean'


signature:
def Rules ( ** rules ) :
    """Create a dictionary mapping symbols to alternative sequences .
 > > > Rules ( A = " B C | D E " )
 { ' A ' : [ [ ' B ' , ' C ' ] , [ ' D ' , ' E ' ] ] }"""

implementation:
for ( lhs , rhs ) in rules . items ( ) :
    rules [ lhs ] = [ alt . strip ( ) . split ( ) for alt in rhs . split ( '|' ) ]
return rules


signature:
def Lexicon ( ** rules ) :
    """Create a dictionary mapping symbols to alternative words .
 > > > Lexicon ( Art = " the | a | an " )
 { ' Art ' : [ ' the ' , ' a ' , ' an ' ] }"""

implementation:
for ( lhs , rhs ) in rules . items ( ) :
    rules [ lhs ] = [ word . strip ( ) for word in rhs . split ( '|' ) ]
return rules


signature:
def generate_random ( grammar = E_ , s = 'S' ) :
    """Replace each token in s by a random entry in grammar ( recursively ) .
 This is useful for testing a grammar , e . g . generate _ random ( E _ )"""

implementation:
import random
def rewrite ( tokens , into ) :
    for token in tokens :
        if token in grammar . rules :
            rewrite ( random . choice ( grammar . rules [ token ] ) , into )
        elif token in grammar . lexicon :
            into . append ( random . choice ( grammar . lexicon [ token ] ) )
        else :
            into . append ( token )
    return into
return ' ' . join ( rewrite ( s . split ( ) , [ ] ) )


signature:
def parses ( self , words , S = 'S' ) :
    """Return a list of parses ; words can be a list or string .
 > > > chart = Chart ( E _ NP _ )
 > > > chart . parses ( ' happy man ' , ' NP ' )
 [ [ 0 , 2 , ' NP ' , [ ( ' Adj ' , ' happy ' ) , [ 1 , 2 , ' NP ' , [ ( ' N ' , ' man ' ) ] , [ ] ] ] , [ ] ] ]"""

implementation:
if isinstance ( words , str ) :
    words = words . split ( )
self . parse ( words , S )
# Return all the parses that span the whole input
# ' span the whole input ' = > begin at 0 , end at len ( words )
return [ [ i , j , S , found , [ ] ] for ( i , j , lhs , found , expects ) in self . chart [ len ( words ) ] # assert j = = len ( words )
if i == 0 and lhs == S and expects == [ ] ]


signature:
def parse ( self , words , S = 'S' ) :
    """Parse a list of words ; according to the grammar .
 Leave results in the chart ."""

implementation:
self . chart = [ [ ] for i in range ( len ( words ) + 1 ) ]
self . add_edge ( [ 0 , 0 , 'S_' , [ ] , [ S ] ] )
for i in range ( len ( words ) ) :
    self . scanner ( i , words [ i ] )
return self . chart


signature:
def add_edge ( self , edge ) :
    "Add edge to chart , and see if it extends or predicts another edge ."

implementation:
start , end , lhs , found , expects = edge
if edge not in self . chart [ end ] :
    self . chart [ end ] . append ( edge )
    if self . trace :
        print '%10s: added %s' % ( caller ( 2 ) , edge )
    if not expects :
        self . extender ( edge )
    else :
        self . predictor ( edge )


signature:
def scanner ( self , j , word ) :
    "For each edge expecting a word of this category here , extend the edge ."

implementation:
for ( i , j , A , alpha , Bb ) in self . chart [ j ] :
    if Bb and self . grammar . isa ( word , Bb [ 0 ] ) :
        self . add_edge ( [ i , j + 1 , A , alpha + [ ( Bb [ 0 ] , word ) ] , Bb [ 1 : ] ] )


signature:
def predictor ( self , ( i , j , A , alpha , Bb ) ) :
    "Add to chart any rules for B that could help extend this edge ."

implementation:
B = Bb [ 0 ]
if B in self . grammar . rules :
    for rhs in self . grammar . rewrites_for ( B ) :
        self . add_edge ( [ j , j , B , [ ] , rhs ] )


signature:
def extender ( self , edge ) :
    "See what edges can be extended by this edge ."

implementation:
( j , k , B , _ , _ ) = edge
for ( i , j , A , alpha , B1b ) in self . chart [ j ] :
    if B1b and B == B1b [ 0 ] :
        self . add_edge ( [ i , k , A , alpha + [ edge ] , B1b [ 1 : ] ] )


signature:
def settings ( request ) :
    """Adds a ` ` SettingDict ` ` object for the ` ` Setting ` ` model to the context as
 ` ` SETTINGS ` ` . Automatically creates non - existent settings with an empty
 string as the default value ."""

implementation:
settings = Setting . objects . all ( ) . as_dict ( default = '' )
context = { 'SETTINGS' : settings , }
return context


signature:
def get_child_models ( self ) :
    """Returns a list of ` ` ( Model , ModelAdmin ) ` ` tuples for ` ` base _ model ` `
 subclasses ."""

implementation:
child_models = [ ]
# Loop through all models with FKs back to ` base _ model ` .
for related_object in get_all_related_objects ( self . base_model . _meta ) : # Django 1.8 deprecated ` get _ all _ related _ objects ( ) ` . We ' re still
# using it for now with the documented work - around for
# compatibility with Django < = 1.7.
    model = getattr ( related_object , 'related_model' , related_object . model )
    # Only consider ` base _ model ` subclasses .
    if issubclass ( model , self . base_model ) :
        class SettingValueAdmin ( self . base_admin_class ) :
            pass
        child_models . append ( ( model , SettingValueAdmin ) )
return child_models


signature:
def DTAgentProgram ( belief_state ) :
    "A decision - theoretic agent . [ Fig . 13.1]"

implementation:
def program ( percept ) :
    belief_state . observe ( program . action , percept )
    program . action = argmax ( belief_state . actions ( ) , belief_state . expected_outcome_utility )
    return program . action
program . action = None
return program


signature:
def event_values ( event , vars ) :
    """Return a tuple of the values of variables vars in event .
 > > > event _ values ( { ' A ' : 10 , ' B ' : 9 , ' C ' : 8 } , [ ' C ' , ' A ' ] )
 (8 , 10)
 > > > event _ values ( ( 1 , 2 ) , [ ' C ' , ' A ' ] )
 (1 , 2)"""

implementation:
if isinstance ( event , tuple ) and len ( event ) == len ( vars ) :
    return event
else :
    return tuple ( [ event [ var ] for var in vars ] )


signature:
def enumerate_joint_ask ( X , e , P ) :
    """Return a probability distribution over the values of the variable X ,
 given the { var : val } observations e , in the JointProbDist P . [ Section 13.3]
 > > > P = JointProbDist ( [ ' X ' , ' Y ' ] )
 > > > P [ 0,0 ] = 0.25 ; P [ 0,1 ] = 0.5 ; P [ 1,1 ] = P [ 2,1 ] = 0.125
 > > > enumerate _ joint _ ask ( ' X ' , dict ( Y = 1 ) , P ) . show _ approx ( )
 '0 : 0.667 , 1 : 0.167 , 2 : 0.167'"""

implementation:
assert X not in e , "Query variable must be distinct from evidence"
Q = ProbDist ( X )
# probability distribution for X , initially empty
Y = [ v for v in P . variables if v != X and v not in e ]
# hidden vars .
for xi in P . values ( X ) :
    Q [ xi ] = enumerate_joint ( Y , extend ( e , X , xi ) , P )
return Q . normalize ( )


signature:
def enumerate_joint ( vars , e , P ) :
    """Return the sum of those entries in P consistent with e ,
 provided vars is P ' s remaining variables ( the ones not in e ) ."""

implementation:
if not vars :
    return P [ e ]
Y , rest = vars [ 0 ] , vars [ 1 : ]
return sum ( [ enumerate_joint ( rest , extend ( e , Y , y ) , P ) for y in P . values ( Y ) ] )


signature:
def enumeration_ask ( X , e , bn ) :
    """Return the conditional probability distribution of variable X
 given evidence e , from BayesNet bn . [ Fig . 14.9]
 > > > enumeration _ ask ( ' Burglary ' , dict ( JohnCalls = T , MaryCalls = T ) , burglary
 . . . ) . show _ approx ( )
 ' False : 0.716 , True : 0.284'"""

implementation:
assert X not in e , "Query variable must be distinct from evidence"
Q = ProbDist ( X )
for xi in bn . variable_values ( X ) :
    Q [ xi ] = enumerate_all ( bn . vars , extend ( e , X , xi ) , bn )
return Q . normalize ( )


signature:
def enumerate_all ( vars , e , bn ) :
    """Return the sum of those entries in P ( vars | e { others } )
 consistent with e , where P is the joint distribution represented
 by bn , and e { others } means e restricted to bn ' s other variables
 ( the ones other than vars ) . Parents must precede children in vars ."""

implementation:
if not vars :
    return 1.0
Y , rest = vars [ 0 ] , vars [ 1 : ]
Ynode = bn . variable_node ( Y )
if Y in e :
    return Ynode . p ( e [ Y ] , e ) * enumerate_all ( rest , e , bn )
else :
    return sum ( Ynode . p ( y , e ) * enumerate_all ( rest , extend ( e , Y , y ) , bn ) for y in bn . variable_values ( Y ) )


signature:
def elimination_ask ( X , e , bn ) :
    """Compute bn ' s P ( X | e ) by variable elimination . [ Fig . 14.11]
 > > > elimination _ ask ( ' Burglary ' , dict ( JohnCalls = T , MaryCalls = T ) , burglary
 . . . ) . show _ approx ( )
 ' False : 0.716 , True : 0.284'"""

implementation:
assert X not in e , "Query variable must be distinct from evidence"
factors = [ ]
for var in reversed ( bn . vars ) :
    factors . append ( make_factor ( var , e , bn ) )
    if is_hidden ( var , X , e ) :
        factors = sum_out ( var , factors , bn )
return pointwise_product ( factors , bn ) . normalize ( )


signature:
def make_factor ( var , e , bn ) :
    """Return the factor for var in bn ' s joint distribution given e .
 That is , bn ' s full joint distribution , projected to accord with e ,
 is the pointwise product of these factors for bn ' s variables ."""

implementation:
node = bn . variable_node ( var )
vars = [ X for X in [ var ] + node . parents if X not in e ]
cpt = dict ( ( event_values ( e1 , vars ) , node . p ( e1 [ var ] , e1 ) ) for e1 in all_events ( vars , bn , e ) )
return Factor ( vars , cpt )


signature:
def sum_out ( var , factors , bn ) :
    "Eliminate var from all factors by summing over its values ."

implementation:
result , var_factors = [ ] , [ ]
for f in factors :
    ( var_factors if var in f . vars else result ) . append ( f )
result . append ( pointwise_product ( var_factors , bn ) . sum_out ( var , bn ) )
return result


signature:
def all_events ( vars , bn , e ) :
    "Yield every way of extending e with values for all vars ."

implementation:
if not vars :
    yield e
else :
    X , rest = vars [ 0 ] , vars [ 1 : ]
    for e1 in all_events ( rest , bn , e ) :
        for x in bn . variable_values ( X ) :
            yield extend ( e1 , X , x )


signature:
def prior_sample ( bn ) :
    """Randomly sample from bn ' s full joint distribution . The result
 is a { variable : value } dict . [ Fig . 14.13]"""

implementation:
event = { }
for node in bn . nodes :
    event [ node . variable ] = node . sample ( event )
return event


signature:
def rejection_sampling ( X , e , bn , N ) :
    """Estimate the probability distribution of variable X given
 evidence e in BayesNet bn , using N samples . [ Fig . 14.14]
 Raises a ZeroDivisionError if all the N samples are rejected ,
 i . e . , inconsistent with e .
 > > > seed ( 47)
 > > > rejection _ sampling ( ' Burglary ' , dict ( JohnCalls = T , MaryCalls = T ) ,
 . . . burglary , 10000 ) . show _ approx ( )
 ' False : 0.7 , True : 0.3'"""

implementation:
counts = dict ( ( x , 0 ) for x in bn . variable_values ( X ) )
# bold N in Fig . 14.14
for j in xrange ( N ) :
    sample = prior_sample ( bn )
    # boldface x in Fig . 14.14
    if consistent_with ( sample , e ) :
        counts [ sample [ X ] ] += 1
return ProbDist ( X , counts )


signature:
def consistent_with ( event , evidence ) :
    "Is event consistent with the given evidence ?"

implementation: return every ( lambda ( k , v ) : evidence . get ( k , v ) == v , event . items ( ) )


signature:
def likelihood_weighting ( X , e , bn , N ) :
    """Estimate the probability distribution of variable X given
 evidence e in BayesNet bn . [ Fig . 14.15]
 > > > seed ( 1017)
 > > > likelihood _ weighting ( ' Burglary ' , dict ( JohnCalls = T , MaryCalls = T ) ,
 . . . burglary , 10000 ) . show _ approx ( )
 ' False : 0.702 , True : 0.298'"""

implementation:
W = dict ( ( x , 0 ) for x in bn . variable_values ( X ) )
for j in xrange ( N ) :
    sample , weight = weighted_sample ( bn , e )
    # boldface x , w in Fig . 14.15
    W [ sample [ X ] ] += weight
return ProbDist ( X , W )


signature:
def weighted_sample ( bn , e ) :
    """Sample an event from bn that ' s consistent with the evidence e ;
 return the event and its weight , the likelihood that the event
 accords to the evidence ."""

implementation:
w = 1
event = dict ( e )
# boldface x in Fig . 14.15
for node in bn . nodes :
    Xi = node . variable
    if Xi in e :
        w *= node . p ( e [ Xi ] , event )
    else :
        event [ Xi ] = node . sample ( event )
return event , w


signature:
def gibbs_ask ( X , e , bn , N ) :
    """[ Fig . 14.16]
 > > > seed ( 1017)
 > > > gibbs _ ask ( ' Burglary ' , dict ( JohnCalls = T , MaryCalls = T ) , burglary , 1000
 . . . ) . show _ approx ( )
 ' False : 0.738 , True : 0.262'"""

implementation:
assert X not in e , "Query variable must be distinct from evidence"
counts = dict ( ( x , 0 ) for x in bn . variable_values ( X ) )
# bold N in Fig . 14.16
Z = [ var for var in bn . vars if var not in e ]
state = dict ( e )
# boldface x in Fig . 14.16
for Zi in Z :
    state [ Zi ] = choice ( bn . variable_values ( Zi ) )
for j in xrange ( N ) :
    for Zi in Z :
        state [ Zi ] = markov_blanket_sample ( Zi , state , bn )
        counts [ state [ X ] ] += 1
return ProbDist ( X , counts )


signature:
def markov_blanket_sample ( X , e , bn ) :
    """Return a sample from P ( X | mb ) where mb denotes that the
 variables in the Markov blanket of X take their values from event
 e ( which must assign a value to each ) . The Markov blanket of X is
 X ' s parents , children , and children ' s parents ."""

implementation:
Xnode = bn . variable_node ( X )
Q = ProbDist ( X )
for xi in bn . variable_values ( X ) :
    ei = extend ( e , X , xi )
    # [ Equation 14.12 : ]
    Q [ xi ] = Xnode . p ( xi , e ) * product ( Yj . p ( ei [ Yj . variable ] , ei ) for Yj in Xnode . children )
return probability ( Q . normalize ( ) [ True ] )


signature:
def normalize ( self ) :
    """Make sure the probabilities of all values sum to 1.
 Returns the normalized distribution .
 Raises a ZeroDivisionError if the sum of the values is 0.
 > > > P = ProbDist ( ' Flip ' ) ; P [ ' H ' ] , P [ ' T ' ] = 35 , 65
 > > > P = P . normalize ( )
 > > > print ' % 5.3f % 5.3f ' % ( P . prob [ ' H ' ] , P . prob [ ' T ' ] )
 0.350 0.650"""

implementation:
total = float ( sum ( self . prob . values ( ) ) )
if not ( 1.0 - epsilon < total < 1.0 + epsilon ) :
    for val in self . prob :
        self . prob [ val ] /= total
return self


signature:
def show_approx ( self , numfmt = '%.3g' ) :
    """Show the probabilities rounded and sorted by key , for the
 sake of portable doctests ."""

implementation: return ', ' . join ( [ ( '%s: ' + numfmt ) % ( v , p ) for ( v , p ) in sorted ( self . prob . items ( ) ) ] )


signature:
def add ( self , node_spec ) :
    """Add a node to the net . Its parents must already be in the
 net , and its variable must not ."""

implementation:
node = BayesNode ( * node_spec )
assert node . variable not in self . vars
assert every ( lambda parent : parent in self . vars , node . parents )
self . nodes . append ( node )
self . vars . append ( node . variable )
for parent in node . parents :
    self . variable_node ( parent ) . children . append ( node )


signature:
def variable_node ( self , var ) :
    """Return the node for the variable named var .
 > > > burglary . variable _ node ( ' Burglary ' ) . variable
 ' Burglary '"""

implementation:
for n in self . nodes :
    if n . variable == var :
        return n
raise Exception ( "No such variable: %s" % var )


signature:
def p ( self , value , event ) :
    """Return the conditional probability
 P ( X = value | parents = parent _ values ) , where parent _ values
 are the values of parents in event . ( event must assign each
 parent a value . )
 > > > bn = BayesNode ( ' X ' , ' Burglary ' , { T : 0.2 , F : 0.625 } )
 > > > bn . p ( False , { ' Burglary ' : False , ' Earthquake ' : True } )
 0.375"""

implementation:
assert isinstance ( value , bool )
ptrue = self . cpt [ event_values ( event , self . parents ) ]
return if_ ( value , ptrue , 1 - ptrue )


signature:
def pointwise_product ( self , other , bn ) :
    "Multiply two factors , combining their variables ."

implementation:
vars = list ( set ( self . vars ) | set ( other . vars ) )
cpt = dict ( ( event_values ( e , vars ) , self . p ( e ) * other . p ( e ) ) for e in all_events ( vars , bn , { } ) )
return Factor ( vars , cpt )


signature:
def sum_out ( self , var , bn ) :
    "Make a factor eliminating var by summing over its values ."

implementation:
vars = [ X for X in self . vars if X != var ]
cpt = dict ( ( event_values ( e , vars ) , sum ( self . p ( extend ( e , var , val ) ) for val in bn . variable_values ( var ) ) ) for e in all_events ( vars , bn , { } ) )
return Factor ( vars , cpt )


signature:
def normalize ( self ) :
    "Return my probabilities ; must be down to one variable ."

implementation:
assert len ( self . vars ) == 1
return ProbDist ( self . vars [ 0 ] , dict ( ( k , v ) for ( ( k , ) , v ) in self . cpt . items ( ) ) )


signature:
def next_chunk_boundaries ( self , buf , prepend_bytes = 0 ) :
    """Computes the next chunk boundaries within ` buf ` .
 See : meth : ` . BaseChunker . next _ chunk _ boundaries ` ."""

implementation: return ( boundary for boundary , _ in self . next_chunk_boundaries_levels ( buf , prepend_bytes ) )


signature:
def next_chunk_boundaries_levels ( self , buf , prepend_bytes = 0 ) :
    """Computes the next chunk boundaries within ` buf ` .
 Similar to : meth : ` . next _ chunk _ boundaries ` , but information about which chunker led to a respective boundary is
 included in the returned value .
 Args :
 buf ( bytes ) : The message that is to be chunked .
 prepend _ bytes ( Optional [ int ] ) : Optional number of zero bytes that should be input to the chunking algorithm
 before ` buf ` .
 Returns :
 list : List of tuples ( boundary , level ) , where boundary is a boundary position relative to ` buf ` and level is
 the index of the chunker ( i . e . , the index of its chunk size specified during instantiation ) that yielded
 the boundary .
 If multiple chunkers yield the same boundary , it is returned only once , along with the highest matching
 chunker index ."""

implementation:
boundaries = { }
for level_index , chunker in enumerate ( self . _chunkers ) :
    boundaries . update ( dict ( [ ( boundary , level_index ) for boundary in chunker . next_chunk_boundaries ( buf , prepend_bytes ) ] ) )
return sorted ( boundaries . items ( ) )


signature:
def create_chunker ( self , chunk_size ) :
    """Create a chunker performing content - defined chunking ( CDC ) using Rabin Karp ' s rolling hash scheme with a
 specific , expected chunk size .
 Args :
 chunk _ size ( int ) : ( Expected ) target chunk size .
 Returns :
 BaseChunker : A chunker object ."""

implementation:
rolling_hash = _rabinkarprh . RabinKarpHash ( self . window_size , self . _seed )
rolling_hash . set_threshold ( 1.0 / chunk_size )
return RabinKarpCDC . _Chunker ( rolling_hash )


signature:
def create_multilevel_chunker ( self , chunk_sizes ) :
    """Create a multi - level chunker performing content - defined chunking ( CDC ) using Rabin Karp ' s rolling hash scheme
 with different specific , expected chunk sizes .
 Args :
 chunk _ sizes ( list ) : List of ( expected ) target chunk sizes .
 Warning :
 For performance reasons , behavior is only defined if chunk sizes are passed in order , i . e . , from
 lowest to highest value .
 Returns :
 BaseMultiLevelChunker : A multi - level chunker object ."""

implementation:
rolling_hash = _rabinkarprh . RabinKarpMultiThresholdHash ( self . window_size , self . _seed , [ 1.0 / chunk_size for chunk_size in chunk_sizes ] )
return RabinKarpCDC . _MultiLevelChunker ( rolling_hash )


signature:
def brightness ( level = 100 , group = 0 ) :
    """Assumes level is out of 100"""

implementation:
if level not in range ( 0 , 101 ) :
    raise Exception ( "Brightness must be value between 0 and 100" )
b = int ( floor ( level / 4.0 ) + 2 )
# lights want values 2-27
return ( COMMANDS [ 'ON' ] [ group ] , Command ( 0x4E , b ) )


signature:
def strip_minidom_whitespace ( node ) :
    """Strips all whitespace from a minidom XML node and its children
 This operation is made in - place ."""

implementation:
for child in node . childNodes :
    if child . nodeType == Node . TEXT_NODE :
        if child . nodeValue :
            child . nodeValue = child . nodeValue . strip ( )
    elif child . nodeType == Node . ELEMENT_NODE :
        strip_minidom_whitespace ( child )


implementation:
if level not in range ( 0 , 101 ) :
    raise Exception ( "Brightness must be value between 0 and 100" )
b = int ( floor ( level / 10.0 ) )
# lights have 10 levels of brightness
commands = list ( darkest ( group ) )
for i in range ( 0 , b ) :
    commands . append ( COMMANDS [ 'BRIGHTER' ] )
return tuple ( commands )


signature:
def warmness ( level = 100 , group = 0 ) :
    """Assumes level is out of 100"""

implementation:
if level not in range ( 0 , 101 ) :
    raise Exception ( "Warmness must be value between 0 and 100" )
b = int ( floor ( level / 10.0 ) )
# lights have 10 levels of warmness
commands = list ( coolest ( group ) )
for i in range ( 0 , b ) :
    commands . append ( COMMANDS [ 'WARMER' ] )
return tuple ( commands )


signature:
def color_from_hls ( hue , light , sat ) :
    """Takes a hls color and converts to proper hue
 Bulbs use a BGR order instead of RGB"""

implementation:
if light > 0.95 : # too bright , let ' s just switch to white
    return 256
elif light < 0.05 : # too dark , let ' s shut it off
    return - 1
else :
    hue = ( - hue + 1 + 2.0 / 3.0 ) % 1
    # invert and translate by 2/3
    return int ( floor ( hue * 256 ) )


signature:
def color_from_rgb ( red , green , blue ) :
    """Takes your standard rgb color
 and converts it to a proper hue value"""

implementation:
r = min ( red , 255 )
g = min ( green , 255 )
b = min ( blue , 255 )
if r > 1 or g > 1 or b > 1 :
    r = r / 255.0
    g = g / 255.0
    b = b / 255.0
return color_from_hls ( * rgb_to_hls ( r , g , b ) )


signature:
def color_from_hex ( value ) :
    """Takes an HTML hex code
 and converts it to a proper hue value"""

implementation:
if "#" in value :
    value = value [ 1 : ]
try :
    unhexed = bytes . fromhex ( value )
except :
    unhexed = binascii . unhexlify ( value )
    # Fallback for 2.7 compatibility
return color_from_rgb ( * struct . unpack ( 'BBB' , unhexed ) )


signature:
def wait ( self , sec = 0.1 ) :
    """Wait for x seconds
 each wait command is 100ms"""

implementation:
sec = max ( sec , 0 )
reps = int ( floor ( sec / 0.1 ) )
commands = [ ]
for i in range ( 0 , reps ) :
    commands . append ( Command ( 0x00 , wait = True ) )
return tuple ( commands )


signature:
def toc ( t = None , name = 'tictoc' ) :
    """ex1)
 tic ( ) # save start time - time1
 toc ( ) # print elapsed time from last calling tic ( )
 toc ( ) # print elapsed time from last calling tic ( )
 ex2)
 t0 = tic ( ) # simple
 t1 = tic ( )
 toc ( t1 ) # print time from t1
 toc ( t0 ) # print time from t0
 : param t : time : 시작 시간 ( tic ( ) 의 리턴 값 )
 : param name : str : 출력시 포함할 문자 [ ' tictoc ' ]"""

implementation:
try :
    t = t or tic . last_tic_time
except AttributeError : # tic ( ) 부터 콜하세요
    logg . warn ( 'calling tic() need to use toc()' )
    return
elapsed = time . time ( ) - t
logg . info ( '%s Elapsed: %s secs' % ( name , elapsed ) )
return elapsed


signature:
def tictoc ( name = 'tictoc' ) :
    """with tictoc ( ' any string or not ' ) :
 print ' cool ~ ~ ~ '
 cool ~ ~ ~
 2015-12-30 14:39:28,458 [ INFO ] tictoc Elapsed : 7.10487365723e - 05 secs
 : param name : str"""

implementation:
t = time . time ( )
yield
logg . info ( '%s Elapsed: %s secs' % ( name , time . time ( ) - t ) )


signature:
def split_rand ( data_or_size , ratio , seed ) :
    """data ( 1 - ratio ) , data ( with ratio ) = split _ rand ( data _ or _ size , ratio , seed )
 : param data _ or _ size : data or count
 : param ratio :
 : param seed :
 : return :"""

implementation:
if not isinstance ( data_or_size , int ) :
    sz = len ( data_or_size )
    data = np . asarray ( data_or_size )
else :
    sz = data_or_size
    data = np . arange ( sz )
if not ratio :
    return data , [ ]
i = np . zeros ( sz , dtype = 'bool' )
lattersz = int ( sz * ratio )
i [ : lattersz ] = True
with np_seed ( seed ) :
    np . random . shuffle ( i )
return data [ ~ i ] , data [ i ]


signature:
def kfolds ( n , k , sz , p_testset = None , seed = 7238 ) :
    """return train , valid [ , test ]
 testset if p _ testset
 : param n :
 : param k :
 : param sz :
 : param p _ testset :
 : param seed :
 : return :"""

implementation:
trains , tests = split_rand ( sz , p_testset , seed )
ntrain = len ( trains )
# np . random . seed ( seed )
with np_seed ( seed ) :
    np . random . shuffle ( trains )
if n == k : # no split
    train , valid = trains , trains
else :
    foldsz = ntrain // k
    itrain = np . arange ( ntrain ) // foldsz != n
    train = trains [ itrain ]
    valid = trains [ ~ itrain ]
if not p_testset :
    return train , valid
else :
    return train , valid , tests


signature:
def date_proc ( func ) :
    """An decorator checking whether date parameter is passing in or not . If not , default date value is all PTT data .
 Else , return PTT data with right date .
 Args :
 func : function you want to decorate .
 request : WSGI request parameter getten from django .
 Returns :
 date :
 a datetime variable , you can only give year , year + month or year + month + day , three type .
 The missing part would be assigned default value 1 ( for month is Jan , for day is 1 ) ."""

implementation:
@ wraps ( func )
def wrapped ( request , * args , ** kwargs ) :
    if 'date' in request . GET and request . GET [ 'date' ] == '' :
        raise Http404 ( "api does not exist" )
    elif 'date' not in request . GET :
        date = datetime . today ( )
        return func ( request , date )
    else :
        date = tuple ( int ( intValue ) for intValue in request . GET [ 'date' ] . split ( '-' ) )
        if len ( date ) == 3 :
            date = datetime ( * date )
        elif len ( date ) == 2 :
            date = datetime ( * date , day = 1 )
        else :
            date = datetime ( * date , month = 1 , day = 1 )
        return func ( request , date )
return wrapped


signature:
def queryString_required ( strList ) :
    """An decorator checking whether queryString key is valid or not
 Args :
 str : allowed queryString key
 Returns :
 if contains invalid queryString key , it will raise exception ."""

implementation:
def _dec ( function ) :
    @ wraps ( function )
    def _wrap ( request , * args , ** kwargs ) :
        for i in strList :
            if i not in request . GET :
                raise Http404 ( "api does not exist" )
        return function ( request , * args , ** kwargs )
    return _wrap
return _dec


signature:
def queryString_required_ClassVersion ( strList ) :
    """An decorator checking whether queryString key is valid or not
 Args :
 str : allowed queryString key
 Returns :
 if contains invalid queryString key , it will raise exception ."""

implementation:
def _dec ( function ) :
    @ wraps ( function )
    def _wrap ( classInstance , request , * args , ** kwargs ) :
        for i in strList :
            if i not in request . GET :
                raise Http404 ( "api does not exist" )
        return function ( classInstance , request , * args , ** kwargs )
    return _wrap
return _dec


signature:
def getJsonFromApi ( view , request ) :
    """Return json from querying Web Api
 Args :
 view : django view function .
 request : http request object got from django .
 Returns : json format dictionary"""

implementation:
jsonText = view ( request )
jsonText = json . loads ( jsonText . content . decode ( 'utf-8' ) )
return jsonText


signature:
def progress ( iter , ** kwargs ) :
    """프로그래스 bar
 for i in progress ( 10 ) :
 print i
 for i in progress ( iter ) :
 print i"""

implementation:
if isinstance ( iter , int ) :
    iter = xrange ( iter )
if hasattr ( iter , '__len__' ) or 'target' in kwargs :
    cls = Progress
else :
    cls = ProgressBase
return cls ( iter , ** kwargs )


signature:
def threaded ( f , * args , ** kwargs ) :
    """function decorator"""

implementation:
if args or kwargs :
    return Threaded ( f , * args , ** kwargs )
@ wraps ( f )
def wrapped ( * wargs , ** wkwargs ) :
    return Threaded ( f , * wargs , ** wkwargs )
return wrapped


signature:
def spawn ( f , * args , ** kwargs ) :
    """decorator"""

implementation:
if args or kwargs :
    return Spawn ( f , * args , ** kwargs )
@ wraps ( f )
def wrapped ( * args , ** kwargs ) :
    return Spawn ( f , * args , ** kwargs )
return wrapped


signature:
def intersect ( self , other ) :
    """self와 other 키가 동일한 아이템의 dictobj
 : type other : dict
 : rtype : dictobj :"""

implementation: return ODict ( ( k , self [ k ] ) for k in self if k in other )


signature:
def from_dict ( dic ) :
    """recursive dict to dictobj 컨버트
 : param dic :
 : return :"""

implementation: return ODict ( ( k , ODict . convert_ifdic ( v ) ) for k , v in dic . items ( ) )


signature:
def plots ( data , ** kwargs ) :
    """simple wrapper plot with labels and skip x
 : param yonly _ or _ xy :
 : param kwargs :
 : return :"""

implementation:
labels = kwargs . pop ( 'labels' , '' )
loc = kwargs . pop ( 'loc' , 1 )
# if len ( yonly _ or _ xy ) = = 1:
# x = range ( len ( yonly _ or _ xy ) )
# y = yonly _ or _ xy
# else :
# x = yonly _ or _ xy [ 0]
# y = yonly _ or _ xy [ 1 : ]
lines = plt . plot ( np . asarray ( data ) . T , ** kwargs )
if labels :
    plt . legend ( lines , labels , loc = loc )
return lines


signature:
def imshow_grid ( images , grid = None , showfun = None , ** opt ) :
    """: param images : nhwc
 : return :"""

implementation:
# assert images . ndim = = 4 or list
showfun = showfun or plt . imshow
count = len ( images )
grid = grid or grid_recommend ( count , sorted ( images [ 0 ] . shape [ : 2 ] ) )
res = [ ]
for i , img in enumerate ( images ) : # grid row first index
    plt . subplot2grid ( grid , ( i % grid [ 0 ] , i // grid [ 0 ] ) )
    res . append ( showfun ( img . squeeze ( ) , ** opt ) )
return res


signature:
def plt_range ( * args , ** kwargs ) :
    """for i in plot _ range ( n ) :
 plt . imshow ( imgs [ i ] )
 left arrow yield prev value
 other key yield next value
 : param args :
 : return :"""

implementation:
wait = kwargs . pop ( 'wait' , True )
if not wait : # no interactive just pass range
    for i in progress ( range ( * args ) ) :
        yield i
    return
class _holder ( object ) :
    pass
hold = _holder ( )
hold . i = 0
hold . done = False
def press ( event ) : # import sys
# sys . stdout . flush ( )
    hold . i += - 1 if event . key == 'left' else 1
    hold . i = 0 if hold . i < 0 else hold . i
def onclose ( event ) :
    hold . done = True
fig = kwargs . pop ( 'fig' , None )
figsize = kwargs . pop ( 'figsize' , None )
if fig is None :
    fig = plt . gcf ( )
    if figsize :
        fig . set_size_inches ( figsize )
elif isinstance ( fig , ( int , str ) ) :
    if figsize :
        fig = plt . figure ( fig , figsize = figsize )
    else :
        fig = plt . figure ( fig )
elif isinstance ( fig , plt . Figure ) :
    if figsize :
        fig . set_size_inches ( figsize )
else :
    raise ValueError
onkey_fig ( press , fig )
onclose_fig ( onclose , fig )
ranges = range ( * args )
l = len ( ranges )
while hold . i < l :
    print ( 'hold.i' , ranges [ hold . i ] )
    yield ranges [ hold . i ]
    # yield first without keypress
    before = hold . i
    while before == hold . i :
        while not fig . waitforbuttonpress ( 0.01 ) :
            if hold . done :
                return
        while fig . waitforbuttonpress ( 0.1 ) :
            if hold . done :
                return


signature:
def plot_pause ( timeout = None , msg = '' ) :
    """todo : add some example
 : param timeout : wait time . if None , blocking
 : param msg :
 : return :"""

implementation:
if timeout is not None :
    print ( msg or 'Press key for continue in time {}' . format ( timeout ) )
    plt . waitforbuttonpress ( timeout = timeout )
    return True
print ( msg or 'Press key for continue' )
while not plt . waitforbuttonpress ( timeout = 0.01 ) :
    if not plt . get_fignums ( ) :
        return False
return len ( plt . get_fignums ( ) ) != 0


signature:
def flat_images ( images , grid = None , bfill = 1.0 , bsz = ( 1 , 1 ) ) :
    """convert batch image to flat image with margin inserted
 [ B , h , w , c ] = > [ H , W , c ]
 : param images :
 : param grid : patch grid cell size of ( Row , Col )
 : param bfill : board filling value
 : param bsz : int or ( int , int ) board size
 : return : flatted image"""

implementation:
if images . ndim == 4 and images . shape [ - 1 ] == 1 :
    images = images . squeeze ( axis = - 1 )
grid = grid or grid_recommend ( len ( images ) , sorted ( images [ 0 ] . shape [ : 2 ] ) )
if not isinstance ( bsz , ( tuple , list ) ) :
    bsz = ( bsz , bsz )
# np . empty ( )
imshape = list ( images . shape )
imshape [ 0 ] = grid [ 0 ] * grid [ 1 ]
imshape [ 1 ] += bsz [ 0 ]
imshape [ 2 ] += bsz [ 1 ]
# data = np . empty ( ( grid [ 0 ] * grid [ 1 ] , imshape [ 1 ] , imshape [ 2 ] ) , dtype = images . dtype )
data = np . empty ( imshape , dtype = images . dtype )
data . fill ( bfill )
bslice0 = slice ( 0 , - bsz [ 0 ] ) if bsz [ 0 ] else slice ( None , None )
bslice1 = slice ( 0 , - bsz [ 1 ] ) if bsz [ 1 ] else slice ( None , None )
data [ : len ( images ) , bslice0 , bslice1 ] = images
imshape = list ( grid ) + imshape [ 1 : ]
# [ grid [ 0 ] , grid [ 1 ] , H , W , [ Channel ] ]
data = data . reshape ( imshape )
if len ( imshape ) == 5 :
    data = data . transpose ( 0 , 2 , 1 , 3 , 4 )
    imshape = [ imshape [ 0 ] * imshape [ 2 ] , imshape [ 1 ] * imshape [ 3 ] , imshape [ 4 ] ]
else : # len = = 4
    data = data . transpose ( 0 , 2 , 1 , 3 )
    imshape = [ imshape [ 0 ] * imshape [ 2 ] , imshape [ 1 ] * imshape [ 3 ] ]
data = data . reshape ( imshape )
# remove last margin
data = data [ bslice0 , bslice1 ]
return data


signature:
def imshow_flat ( images , grid = None , showfun = None , bfill = 1.0 , bsz = ( 1 , 1 ) , ** opt ) :
    """imshow after applying flat _ images
 : param images : [ bhwc ]
 : param grid : None for auto grid
 : param showfun : plt . imshow
 : param bfill : color for board fill
 : param bsz : size of board
 : param opt : option for showfun
 : return :"""

implementation:
showfun = showfun or plt . imshow
count = len ( images )
# decide grid shape if need pick one
grid = grid or grid_recommend ( count , ratio = sorted ( images [ 0 ] . shape [ : 2 ] ) )
flatted = flat_images ( images , grid , bfill = bfill , bsz = bsz )
res = showfun ( flatted , ** opt )
plt . draw ( )


signature:
def matshow ( * args , ** kwargs ) :
    """imshow without interpolation like as matshow
 : param args :
 : param kwargs :
 : return :"""

implementation:
kwargs [ 'interpolation' ] = kwargs . pop ( 'interpolation' , 'none' )
return plt . imshow ( * args , ** kwargs )


signature:
def imbox ( xy , w , h , angle = 0.0 , ** kwargs ) :
    """draw boundary box
 : param xy : start index xy ( ji )
 : param w : width
 : param h : height
 : param angle :
 : param kwargs :
 : return :"""

implementation:
from matplotlib . patches import Rectangle
return imbound ( Rectangle , xy , w , h , angle , ** kwargs )


signature:
def imbound ( clspatch , * args , ** kwargs ) :
    """: param clspatch :
 : param args :
 : param kwargs :
 : return :"""

implementation:
# todo : add example
c = kwargs . pop ( 'color' , kwargs . get ( 'edgecolor' , None ) )
kwargs . update ( facecolor = 'none' , edgecolor = c )
return impatch ( clspatch , * args , ** kwargs )


signature:
def imslic ( img , n_segments = 100 , aspect = None ) :
    """slic args :
 n _ segments = 100 , compactness = 10 . , max _ iter = 10,
 sigma = 0 , spacing = None ,
 multichannel = True , convert2lab = None , enforce _ connectivity = True ,
 min _ size _ factor = 0.5 , max _ size _ factor = 3 , slic _ zero = False
 mark _ boundaries args :
 label _ img , color = ( 1 , 1 , 0 ) , outline _ color = None , mode = ' outer ' , background _ label = 0
 imshow args :
 cmap = None , norm = None , aspect = None , interpolation = None ,
 alpha = None , vmin = None , vmax = None , origin = None ,
 extent = None , shape = None , filternorm = 1 , filterrad = 4.0,
 imlim = None , resample = None , url = None , hold = None , data = None ,
 : param img :
 : param slicarg :
 : param slickw :
 : return :"""

implementation:
from skimage . segmentation import ( slic , mark_boundaries )
from skimage . morphology import ( dilation )
if img . ndim == 2 or img . ndim == 3 and img . shape [ - 1 ] == 1 :
    imz = np . stack ( [ img , img , img ] , 2 )
else :
    imz = img
slics = slic ( imz , n_segments = n_segments )
boundaries = mark_boundaries ( imz , slics )
return plt . imshow ( boundaries , aspect = aspect )


signature:
def imslic2 ( img , n_segments = 100 , color = None , outline_color = None , mode = 'thick' , ** kwargs ) :
    """slic args :
 n _ segments = 100 , compactness = 10 . , max _ iter = 10,
 sigma = 0 , spacing = None ,
 multichannel = True , convert2lab = None , enforce _ connectivity = True ,
 min _ size _ factor = 0.5 , max _ size _ factor = 3 , slic _ zero = False
 mark _ boundaries args :
 label _ img , color = ( 1 , 1 , 0 ) , outline _ color = None , mode = ' outer ' , background _ label = 0
 imshow args :
 cmap = None , norm = None , aspect = None , interpolation = None ,
 alpha = None , vmin = None , vmax = None , origin = None ,
 extent = None , shape = None , filternorm = 1 , filterrad = 4.0,
 imlim = None , resample = None , url = None , hold = None , data = None ,
 : param img :
 : param slicarg :
 : param slickw :
 : return :"""

implementation:
from skimage . segmentation import ( slic , find_boundaries )
# mark _ boundaries
from skimage . morphology import ( dilation )
kwslic = { 'compactness' , 'max_iter' , 'sigma' , 'spacing' , 'multichannel' , 'convert2lab' , 'enforce_connectivity' , 'min_size_factor' , 'max_size_factor' , 'slic_zero=False' }
imshowkw = { 'cmap' , 'norm' , 'aspect' , 'interpolation' , 'alpha' , 'vmin' , 'vmax' , 'origin' , 'extent' , 'shape' , 'filternorm' , 'filterrad' , 'imlim' , 'resample' , 'url' , 'hold' , 'data' }
slicarg = { k : v for k , v in kwargs . iteritems ( ) if k in kwslic }
imshowarg = { k : v for k , v in kwargs . iteritems ( ) if k in imshowkw }
if img . ndim == 2 or img . ndim == 3 and img . shape [ - 1 ] == 1 :
    imz = np . stack ( [ img , img , img ] , 2 )
    color = color or 1.
else :
    imgz = img
    color = color or ( 1 , 1 , 0 )
slics = slic ( imz , n_segments = n_segments , ** slicarg )
boundaries = find_boundaries ( slics , mode = mode )
if outline_color is not None :
    outlines = dilation ( boundaries , np . ones ( ( 3 , 3 ) , np . uint8 ) )
    img [ outlines ] = outline_color
img [ boundaries ] = color
return plt . imshow ( img , ** imshowarg )


signature:
def movie_saving ( outfile , showfun = imshow , fig = None , tight = True , drawopt = None , dpi = 100 , ** movieopt ) :
    """contextmanager for PlotMovieWriter
 Example :
 with movie _ saving ( ' output . mp4 ' , dpi = 100 ) as plot :
 for i in range ( 10 ) :
 plot ( data [ i ] )
 : param outfile :
 : param showfun :
 : param fig :
 : param tight :
 : param drawopt :
 : param dpi :
 : param movieopt : fps = 5 , codec = None , bitrate = None , extra _ args = None , metadata = None
 : return :"""

implementation:
if tight :
    plot_writer = ImageMovieWriter ( outfile , showfun = showfun , fig = fig , drawopt = drawopt , dpi = dpi , ** movieopt )
else :
    plot_writer = PlotMovieWriter ( outfile , showfun = showfun , fig = fig , drawopt = drawopt , dpi = dpi , ** movieopt )
try :
    yield plot_writer
finally :
    plot_writer . finish ( )


signature:
def put ( xy , * args ) :
    """put text on on screen
 a tuple as first argument tells absolute position for the text
 does not change TermCursor position
 args = list of optional position , formatting tokens and strings"""

implementation:
cmd = [ TermCursor . save , TermCursor . move ( * xy ) , '' . join ( args ) , TermCursor . restore ]
write ( '' . join ( cmd ) )


signature:
def getpassword ( prompt = "Password: " ) :
    """get user input without echo"""

implementation:
fd = sys . stdin . fileno ( )
old = termios . tcgetattr ( fd )
new = termios . tcgetattr ( fd )
new [ 3 ] &= ~ termios . ECHO
# lflags
try :
    termios . tcsetattr ( fd , termios . TCSADRAIN , new )
    passwd = raw_input ( prompt )
finally :
    termios . tcsetattr ( fd , termios . TCSADRAIN , old )
return passwd


signature:
def getch ( ) :
    """get character . waiting for key"""

implementation:
try :
    termios . tcsetattr ( _fd , termios . TCSANOW , _new_settings )
    ch = sys . stdin . read ( 1 )
finally :
    termios . tcsetattr ( _fd , termios . TCSADRAIN , _old_settings )
return ch


signature:
def getlogger ( pkg = '' , handler = None ) :
    """패키지 혹은 채널 로거
 logging . getLogger ( package _ name ) or logg . getLogger ( )
 : param pkg : str"""

implementation:
from . caller import caller
if not pkg :
    m = caller . modulename ( )
    s = m . split ( '.' , 1 )
    if len ( s ) > 1 :
        pkg = s [ 0 ]
if haslogger ( pkg ) :
    return logging . getLogger ( pkg )
else : # local
    logger = logging . getLogger ( pkg )
    logger . addHandler ( handler or default_handler )
    logger . setLevel ( logging . DEBUG )
    return logger


signature:
def basicConfig ( ** kw ) :
    """logging의 로그를 한번 호출하면 basicConfig가 안먹으므로 . 기존 핸들러 삭제후 재설정 .
 http : / / stackoverflow . com / questions / 1943747 / python - logging - before - you - run - logging - basicconfig
 ex )
 basicConfig ( format = ' % ( asctime ) s % ( message ) s ' , level = logging . DEBUG )
 : param filename : Specifies that a FileHandler be created , using the specified filename , rather than a StreamHandler .
 : param filemode : Specifies the mode to open the file , if filename is specified ( if filemode is unspecified , it defaults to ‘ a ’ ) .
 : param format : Use the specified format string for the handler . ( https : / / docs . python . org / 2.7 / library / logging . html # logging . basicConfig
 : param datefmt : Use the specified date / time format .
 : param level : Set the root logger level to the specified level .
 : param stream : Use the specified stream to initialize the StreamHandler . Note that this argument is incompatible with ‘ filename ’ - if both are present , ‘ stream ’ is ignored ."""

implementation:
while len ( logging . root . handlers ) > 0 :
    logging . root . removeHandler ( logging . root . handlers [ - 1 ] )
logging . basicConfig ( ** kw )


signature:
def format ( self , record ) :
    """tweaked from source of base"""

implementation:
try :
    record . message = record . getMessage ( )
except TypeError : # if error during msg = msg % self . args
    if record . args :
        if isinstance ( record . args , collections . Mapping ) :
            record . message = record . msg . format ( ** record . args )
        else :
            record . message = record . msg . format ( record . args )
self . _fmt = self . getfmt ( record . levelname )
if self . usesTime ( ) :
    record . asctime = self . formatTime ( record , self . datefmt )
s = self . _fmt . format ( ** record . __dict__ )
if record . exc_info : # Cache the traceback text to avoid converting it multiple times
# ( it ' s constant anyway )
    if not record . exc_text :
        record . exc_text = self . formatException ( record . exc_info )
if record . exc_text :
    if s [ - 1 : ] != '\n' :
        s += '\n'
    try :
        s = s + record . exc_text
    except UnicodeError :
        s = s + record . exc_text . decode ( sys . getfilesystemencoding ( ) , 'replace' )
return s


signature:
def getProcessOwner ( pid ) :
    '''getProcessOwner - Get the process owner of a pid
 @ param pid < int > - process id
 @ return - None if process not found or can ' t be determined . Otherwise , a dict :
 uid - Owner UID
 name - Owner name , or None if one cannot be determined'''

implementation:
try :
    ownerUid = os . stat ( '/proc/' + str ( pid ) ) . st_uid
except :
    return None
try :
    ownerName = pwd . getpwuid ( ownerUid ) . pw_name
except :
    ownerName = None
return { 'uid' : ownerUid , 'name' : ownerName }


signature:
def getProcessOwnerStr ( pid ) :
    '''getProcessOwner - Get Process owner of a pid as a string instead of components ( # getProcessOwner )
 @ return - Returns username if it can be determined , otherwise uid , otherwise " unknown "'''

implementation:
ownerInfo = getProcessOwner ( pid )
if ownerInfo :
    if ownerInfo [ 'name' ] :
        owner = ownerInfo [ 'name' ]
    else :
        owner = str ( ownerInfo [ 'uid' ] )
else :
    owner = 'unknown'
return owner


signature:
def getProcessCommandLineStr ( pid ) :
    '''getProcessCommandLineStr - Gets a the commandline ( program + arguments ) of a given pid
 @ param pid < int > - Process ID
 @ return - None if process not found or can ' t be determined . Otherwise a string of commandline .
 @ note Caution , args may have spaces in them , and you cannot surmise from this method . If you care ( like trying to replay a command ) , use getProcessCommandLineList instead'''

implementation:
try :
    with open ( '/proc/%d/cmdline' % ( int ( pid ) , ) , 'r' ) as f :
        cmdline = f . read ( )
    return cmdline . replace ( '\x00' , ' ' )
except :
    return None


signature:
def getProcessCommandLineList ( pid ) :
    '''getProcessCommandLineList - Gets the commandline ( program + argumentS ) of a given pid as a list .
 @ param pid < int > - Process ID
 @ return - None if process not found or can ' t be determined . Otherwise a list representing argv . First argument is process name , remainder are arguments .
 @ note - Use this if you care about whether a process had a space in the commands'''

implementation:
try :
    with open ( '/proc/%d/cmdline' % ( int ( pid ) , ) , 'r' ) as f :
        cmdline = f . read ( )
    return cmdline . split ( '\x00' )
except :
    return None


signature:
def scanProcessForCwd ( pid , searchPortion , isExactMatch = False ) :
    '''scanProcessForCwd - Searches a given pid ' s cwd for a given pattern
 @ param pid < int > - A running process ID on this system
 @ param searchPortion < str > - Any portion of directory to search
 @ param isExactMatch < bool > Default False - If match should be exact , otherwise a partial match is performed .
 @ return < dict > - If result is found , the following dict is returned . If no match found on the given pid , or pid is not found running , None is returned .
 ' searchPortion ' : The passed search pattern
 ' pid ' : The passed pid ( as an integer )
 ' owner ' : String of process owner , or uid if no mapping can be found , or " unknown " if neither could be determined .
 ' cmdline ' : Commandline string
 ' cwd ' : The exact cwd of matched process'''

implementation:
try :
    try :
        pid = int ( pid )
    except ValueError as e :
        sys . stderr . write ( 'Expected an integer, got %s for pid.\n' % ( str ( type ( pid ) ) , ) )
        raise e
    cwd = getProcessCwd ( pid )
    if not cwd :
        return None
    isMatch = False
    if isExactMatch is True :
        if searchPortion == cwd :
            isMatch = True
        else :
            if searchPortion . endswith ( '/' ) and searchPortion [ : - 1 ] == cwd :
                isMatch = True
    else :
        if searchPortion in cwd :
            isMatch = True
        else :
            if searchPortion . endswith ( '/' ) and searchPortion [ : - 1 ] in cwd :
                isMatch = True
    if not isMatch :
        return None
    cmdline = getProcessCommandLineStr ( pid )
    owner = getProcessOwnerStr ( pid )
    return { 'searchPortion' : searchPortion , 'pid' : pid , 'owner' : owner , 'cmdline' : cmdline , 'cwd' : cwd , }
except OSError :
    return None
except IOError :
    return None
except FileNotFoundError :
    return None
except PermissionError :
    return None


signature:
def scanAllProcessesForCwd ( searchPortion , isExactMatch = False ) :
    '''scanAllProcessesForCwd - Scans all processes on the system for a given search pattern .
 @ param searchPortion < str > - Any portion of directory to search
 @ param isExactMatch < bool > Default False - If match should be exact , otherwise a partial match is performed .
 @ return - < dict > - A dictionary of pid - > cwdResults for each pid that matched the search pattern . For format of " cwdResults " , @ see scanProcessForCwd'''

implementation:
pids = getAllRunningPids ( )
cwdResults = [ scanProcessForCwd ( pid , searchPortion , isExactMatch ) for pid in pids ]
ret = { }
for i in range ( len ( pids ) ) :
    if cwdResults [ i ] is not None :
        ret [ pids [ i ] ] = cwdResults [ i ]
return ret


signature:
def scanProcessForMapping ( pid , searchPortion , isExactMatch = False , ignoreCase = False ) :
    '''scanProcessForMapping - Searches a given pid ' s mappings for a certain pattern .
 @ param pid < int > - A running process ID on this system
 @ param searchPortion < str > - A mapping for which to search , example : libc or python or libz . so . 1 . Give empty string to return all mappings .
 @ param isExactMatch < bool > Default False - If match should be exact , otherwise a partial match is performed .
 @ param ignoreCase < bool > Default False - If True , search will be performed case - insensitively
 @ return < dict > - If result is found , the following dict is returned . If no match found on the given pid , or pid is not found running , None is returned .
 ' searchPortion ' : The passed search pattern
 ' pid ' : The passed pid ( as an integer )
 ' owner ' : String of process owner , or uid if no mapping can be found , or " unknown " if neither could be determined .
 ' cmdline ' : Commandline string
 ' matchedMappings ' : All mappings likes that matched the given search pattern'''

implementation:
try :
    try :
        pid = int ( pid )
    except ValueError as e :
        sys . stderr . write ( 'Expected an integer, got %s for pid.\n' % ( str ( type ( pid ) ) , ) )
        raise e
    with open ( '/proc/%d/maps' % ( pid , ) , 'r' ) as f :
        contents = f . read ( )
    lines = contents . split ( '\n' )
    matchedMappings = [ ]
    if isExactMatch is True :
        if ignoreCase is False :
            isMatch = lambda searchFor , searchIn : bool ( searchFor == searchIn )
        else :
            isMatch = lambda searchFor , searchIn : bool ( searchFor . lower ( ) == searchIn . lower ( ) )
    else :
        if ignoreCase is False :
            isMatch = lambda searchFor , searchIn : bool ( searchFor in searchIn )
        else :
            isMatch = lambda searchFor , searchIn : bool ( searchFor . lower ( ) in searchIn . lower ( ) )
    for line in lines :
        portion = ' ' . join ( line . split ( ' ' ) [ 5 : ] ) . lstrip ( )
        if isMatch ( searchPortion , portion ) :
            matchedMappings . append ( '\t' + line )
    if len ( matchedMappings ) == 0 :
        return None
    cmdline = getProcessCommandLineStr ( pid )
    owner = getProcessOwnerStr ( pid )
    return { 'searchPortion' : searchPortion , 'pid' : pid , 'owner' : owner , 'cmdline' : cmdline , 'matchedMappings' : matchedMappings , }
except OSError :
    return None
except IOError :
    return None
except FileNotFoundError :
    return None
except PermissionError :
    return None


signature:
def scanAllProcessesForMapping ( searchPortion , isExactMatch = False , ignoreCase = False ) :
    '''scanAllProcessesForMapping - Scans all processes on the system for a given search pattern .
 @ param searchPortion < str > - A mapping for which to search , example : libc or python or libz . so . 1 . Give empty string to return all mappings .
 @ param isExactMatch < bool > Default False - If match should be exact , otherwise a partial match is performed .
 @ param ignoreCase < bool > Default False - If True , search will be performed case - insensitively
 @ return - < dict > - A dictionary of pid - > mappingResults for each pid that matched the search pattern . For format of " mappingResults " , @ see scanProcessForMapping'''

implementation:
pids = getAllRunningPids ( )
# Since processes could disappear , we run the scan as fast as possible here with a list comprehension , then assemble the return dictionary later .
mappingResults = [ scanProcessForMapping ( pid , searchPortion , isExactMatch , ignoreCase ) for pid in pids ]
ret = { }
for i in range ( len ( pids ) ) :
    if mappingResults [ i ] is not None :
        ret [ pids [ i ] ] = mappingResults [ i ]
return ret


signature:
def scanProcessForOpenFile ( pid , searchPortion , isExactMatch = True , ignoreCase = False ) :
    '''scanProcessForOpenFile - Scans open FDs for a given pid to see if any are the provided searchPortion
 @ param searchPortion < str > - Filename to check
 @ param isExactMatch < bool > Default True - If match should be exact , otherwise a partial match is performed .
 @ param ignoreCase < bool > Default False - If True , search will be performed case - insensitively
 @ return - If result is found , the following dict is returned . If no match found on the given pid , or the pid is not found running , None is returned .
 ' searchPortion ' : The search portion provided
 ' pid ' : The passed pid ( as an integer )
 ' owner ' : String of process owner , or " unknown " if one could not be determined
 ' cmdline ' : Commandline string
 ' fds ' : List of file descriptors assigned to this file ( could be mapped several times )
 ' filenames ' : List of the filenames matched'''

implementation:
try :
    try :
        pid = int ( pid )
    except ValueError as e :
        sys . stderr . write ( 'Expected an integer, got %s for pid.\n' % ( str ( type ( pid ) ) , ) )
        raise e
    prefixDir = "/proc/%d/fd" % ( pid , )
    processFDs = os . listdir ( prefixDir )
    matchedFDs = [ ]
    matchedFilenames = [ ]
    if isExactMatch is True :
        if ignoreCase is False :
            isMatch = lambda searchFor , totalPath : bool ( searchFor == totalPath )
        else :
            isMatch = lambda searchFor , totalPath : bool ( searchFor . lower ( ) == totalPath . lower ( ) )
    else :
        if ignoreCase is False :
            isMatch = lambda searchFor , totalPath : bool ( searchFor in totalPath )
        else :
            isMatch = lambda searchFor , totalPath : bool ( searchFor . lower ( ) in totalPath . lower ( ) )
    for fd in processFDs :
        fdPath = os . readlink ( prefixDir + '/' + fd )
        if isMatch ( searchPortion , fdPath ) :
            matchedFDs . append ( fd )
            matchedFilenames . append ( fdPath )
    if len ( matchedFDs ) == 0 :
        return None
    cmdline = getProcessCommandLineStr ( pid )
    owner = getProcessOwnerStr ( pid )
    return { 'searchPortion' : searchPortion , 'pid' : pid , 'owner' : owner , 'cmdline' : cmdline , 'fds' : matchedFDs , 'filenames' : matchedFilenames , }
except OSError :
    return None
except IOError :
    return None
except FileNotFoundError :
    return None
except PermissionError :
    return None


signature:
def scanAllProcessesForOpenFile ( searchPortion , isExactMatch = True , ignoreCase = False ) :
    '''scanAllProcessessForOpenFile - Scans all processes on the system for a given filename
 @ param searchPortion < str > - Filename to check
 @ param isExactMatch < bool > Default True - If match should be exact , otherwise a partial match is performed .
 @ param ignoreCase < bool > Default False - If True , search will be performed case - insensitively
 @ return - < dict > - A dictionary of pid - > mappingResults for each pid that matched the search pattern . For format of " mappingResults " , @ see scanProcessForOpenFile'''

implementation:
pids = getAllRunningPids ( )
# Since processes could disappear , we run the scan as fast as possible here with a list comprehension , then assemble the return dictionary later .
mappingResults = [ scanProcessForOpenFile ( pid , searchPortion , isExactMatch , ignoreCase ) for pid in pids ]
ret = { }
for i in range ( len ( pids ) ) :
    if mappingResults [ i ] is not None :
        ret [ pids [ i ] ] = mappingResults [ i ]
return ret


signature:
def enum ( name , * members , ** withvalue ) :
    """class buider"""

implementation:
if len ( members ) == 1 :
    if isinstance ( members [ 0 ] , str ) :
        members = members [ 0 ] . split ( )
    elif isinstance ( members [ 0 ] , ( list , tuple ) ) :
        members = members [ 0 ]
dic = { v : v for v in members }
dic . update ( withvalue )
return type ( name , ( Enum , ) , dic )


signature:
def database ( db = '' , ** kwargs ) :
    """usage :
 with database ( ' my _ db ' ) as conn :
 c = conn . cursor ( )
 database 커넥션 with 문과 같이 사용하고 , 알아서 close하기
 : param db : str : db스키마
 : param kwargs :
 : return :"""

implementation:
db = kwargs . pop ( 'db' , db )
arg = db_config ( db )
arg . update ( kwargs )
return closing ( MySQLdb . connect ( ** arg ) )


signature:
def connect ( db = '' , ** kwargs ) :
    """db 접속 공통 인자들 채워서 접속 , schema만 넣으면 됩니다 .
 db connection 객체 반환이지만
 with 문과 같이 쓰이면 cursor임에 주의 ( MySQLdb의 구현이 그렇습니다 . )
 ex1)
 import snipy . database as db
 conn = db . connect ( ' my _ db ' )
 cursor = conn . cursor ( )
 ex2)
 import snipy . database as db
 with db . connect ( ' my _ db ' ) as cursor :
 cursor . execute ( query )
 : param db : str : db schema
 : param kwargs : 추가 접속 정보
 : return : connection or cursor"""

implementation:
arg = db_config ( db )
arg . update ( kwargs )
return MySQLdb . connect ( ** arg )


signature:
def _cursor_exit ( cursor , exc_type , exc_value , traceback ) :
    """cursor with문과 쓸수 있게 _ _ exit _ _ 에 바인딩
 : param cursor :
 : param exc _ type :
 : param exc _ value :
 : param traceback :
 : return :"""

implementation:
if exc_type is not None :
    print ( exc_value , traceback )
cursor . connection . close ( )


signature:
def fetch ( query , args = None , ** kwargs ) :
    """for record in fetch ( query , args , * * configs ) :
 print record
 : param args :
 : param db : str : db 스키마
 : param query : 쿼리 스트링
 : param kwargs : db connection 추가 인자 . 보통 생략
 : return : iterator"""

implementation:
cur = execute ( kwargs . pop ( 'db' , '' ) , query , args , ** kwargs )
for r in cur :
    yield r
cur . connection . close ( )


signature:
def get_insert_query ( table , fields = None , field_count = None ) :
    """format insert query
 : param table : str
 : param fields : list [ str ]
 : param field _ count : int
 : return : str"""

implementation:
if fields :
    q = 'insert into %s ({0}) values ({1});' % table
    l = len ( fields )
    q = q . format ( ',' . join ( fields ) , ',' . join ( [ '%s' ] * l ) )
elif field_count :
    q = 'insert into %s values ({0});' % table
    q = q . format ( ',' . join ( [ '%s' ] * field_count ) )
else :
    raise ValueError ( 'fields or field_count need' )
return q


signature:
def insert ( cursor , table , * args , ** field_values ) :
    """db에 레코드 집어넣기
 ex )
 cursor . insert ( table , v1 , v2 , . . . )
 ex )
 cursor . insert ( table , id = v1 , word = v2 , commit = True )
 : param commit :
 : param cursor :
 : param table :
 : param args :
 : param field _ values :
 : return :"""

implementation:
commit = field_values . pop ( 'commit' , True )
q , a = None , None
if args is not None and len ( args ) > 0 :
    q = get_insert_query ( table , field_count = len ( args ) )
    a = args
elif len ( field_values ) > 0 :
    q = get_insert_query ( table , fields = field_values . keys ( ) )
    a = field_values . values ( )
else :
    raise ValueError ( 'need table, record...' )
cursor . execute ( q , args = a )
if commit :
    cursor . connection . commit ( )


signature:
def update ( cursor , table , where_kv , commit = True , ** field_values ) :
    """db update 쿼리 빌딩 및 실행 , 단 , commit은
 : param cursor : 커서
 : type cursor : Cursor
 : param table : 테이블 이름
 : type table : str
 : param where _ kv : 업데이트 where 조건 dictionary , key : field , value : equal condition only
 : type where _ kv : dict
 : param field _ values : kwarg 업데이트용
 : type field _ values : dict
 : param commit : 커밋 여부
 : type commit : bool
 : return :"""

implementation:
q = """update %s \nset {0} \nwhere {1} """ % table
fields = field_values . keys ( )
kv = ',' . join ( [ '{}=%s' . format ( f ) for f in fields ] )
where = ' and ' . join ( [ '{}=%s' . format ( f ) for f in where_kv . keys ( ) ] )
q = q . format ( kv , where )
args = field_values . values ( ) + where_kv . values ( )
cursor . execute ( q , args = args )
if commit :
    cursor . connection . commit ( )


signature:
def insert_or_update ( cursor , table , commit = True , ** field_values ) :
    """db update 쿼리 빌딩 및 실행 , 단 , commit은
 : param cursor : 커서
 : type cursor : Cursor
 : param table : 테이블이름
 : type table : str
 : param commit : 커밋 여부
 : type commit : bool
 : param field _ values : insert 또는 업데이트 할 필드 및 값 dict pairs
 : type field _ values : dict
 : return :"""

implementation:
q = """INSERT INTO %s ({0}) \nVALUES ({1}) \nON DUPLICATE KEY UPDATE {2} """ % table
l = len ( field_values )
fields = field_values . keys ( )
field = ',' . join ( fields )
value = ',' . join ( [ '%s' ] * l )
kv = ',' . join ( [ '{}=%s' . format ( f ) for f in fields ] )
q = q . format ( field , value , kv )
args = field_values . values ( ) * 2
cursor . execute ( q , args = args )
if commit :
    cursor . connection . commit ( )


signature:
def tojson ( o ) :
    """recursive implementation"""

implementation:
try :
    return json . encode ( o )
except json . EncodeError :
    pass
try :
    return o . tojson ( )
except AttributeError as e :
    pass
t = type ( o )
if isinstance ( o , list ) :
    return '[%s]' % ', ' . join ( [ tojson ( e ) for e in o ] )
elif isinstance ( o , dict ) :
    d = [ '%s:%s' % ( k , tojson ( v ) ) for k , v in o . iteritems ( ) ]
    return '{%s}' % ', ' . join ( d )
elif isinstance ( o , set ) :
    d = [ '%s:%s' % ( tojson ( e ) ) for e in o ]
    return '{%s}' % ', ' . join ( d )
elif isinstance ( o , np . ndarray ) :
    return numpy_to_json ( o )
else :
    raise ValueError ( 'error, failed encoding type(%s) to json' % t )


signature:
def named ( typename , * fieldnames , ** defaults ) :
    """namedtuple with default values
 named ( ' typename ' , fields | * fields , default = x , [ * * defaults ] )
 : param typename :
 : param fieldnames :
 : param defaults :
 : return :"""

implementation:
if len ( fieldnames ) == 1 :
    if isinstance ( fieldnames [ 0 ] , str ) :
        fieldnames = tuple ( fieldnames [ 0 ] . replace ( ',' , ' ' ) . split ( ) )
    elif isinstance ( fieldnames [ 0 ] , ( list , tuple ) ) :
        fieldnames = fieldnames [ 0 ]
# set default of defaults
default_of_defaults = defaults . pop ( 'default' , None )
dfields = tuple ( f for f in defaults if f not in fieldnames )
T = collections . namedtuple ( typename , fieldnames + dfields )
T . __new__ . __defaults__ = ( default_of_defaults , ) * len ( T . _fields )
prototype = T ( ** defaults )
T . __new__ . __defaults__ = tuple ( prototype )
# make picklable
globals ( ) [ typename ] = T
return T


signature:
def np_seed ( seed ) :
    """numpy random seed context
 : param seed :
 : return :"""

implementation:
if seed is not None :
    state = np . random . get_state ( )
    np . random . seed ( seed )
    yield
    np . random . set_state ( state )
else :
    yield


signature:
def connect ( self ) :
    """Create and connect to socket for TCP communication with hub ."""

implementation:
try :
    self . _socket = socket . socket ( socket . AF_INET , socket . SOCK_STREAM )
    self . _socket . settimeout ( TIMEOUT_SECONDS )
    self . _socket . connect ( ( self . _ip , self . _port ) )
    _LOGGER . debug ( "Successfully created Hub at %s:%s :)" , self . _ip , self . _port )
except socket . error as error :
    _LOGGER . error ( "Error creating Hub: %s :(" , error )
    self . _socket . close ( )


signature:
def send_command ( self , command ) :
    """Send TCP command to hub and return response ."""

implementation:
# use lock to make TCP send / receive thread safe
with self . _lock :
    try :
        self . _socket . send ( command . encode ( "utf8" ) )
        result = self . receive ( )
        # hub may send " status " / " new " messages that should be ignored
        while result . startswith ( "S" ) or result . startswith ( "NEW" ) :
            _LOGGER . debug ( "!Got response: %s" , result )
            result = self . receive ( )
        _LOGGER . debug ( "Received: %s" , result )
        return result
    except socket . error as error :
        _LOGGER . error ( "Error sending command: %s" , error )
        # try re - connecting socket
        self . connect ( )
        return ""