signature:
def _make_signature_key ( args , kwargs ) :
    """Transforms function args into a key that can be used by the cache
 CommandLine :
 xdoctest - m ubelt . util _ memoize _ make _ signature _ key
 Example :
 > > > args = ( 4 , [ 1 , 2 ] )
 > > > kwargs = { ' a ' : ' b ' }
 > > > key = _ make _ signature _ key ( args , kwargs )
 > > > print ( ' key = { ! r } ' . format ( key ) )
 > > > # Some mutable types cannot be handled by ub . hash _ data
 > > > import pytest
 > > > import six
 > > > if six . PY2:
 > > > import collections as abc
 > > > else :
 > > > from collections import abc
 > > > with pytest . raises ( TypeError ) :
 > > > _ make _ signature _ key ( ( 4 , [ 1 , 2 ] , { 1 : 2 , ' a ' : ' b ' } ) , kwargs = { } )
 > > > class Dummy ( abc . MutableSet ) :
 > > > def _ _ contains _ _ ( self , item ) : return None
 > > > def _ _ iter _ _ ( self ) : return iter ( [ ] )
 > > > def _ _ len _ _ ( self ) : return 0
 > > > def add ( self , item , loc ) : return None
 > > > def discard ( self , item ) : return None
 > > > with pytest . raises ( TypeError ) :
 > > > _ make _ signature _ key ( ( Dummy ( ) , ) , kwargs = { } )"""

implementation:
kwitems = kwargs . items ( )
# TODO : we should check if Python is at least 3.7 and sort by kwargs
# keys otherwise . Should we use hash _ data for key generation
if ( sys . version_info . major , sys . version_info . minor ) < ( 3 , 7 ) : # nocover
# We can sort because they keys are gaurenteed to be strings
    kwitems = sorted ( kwitems )
kwitems = tuple ( kwitems )
try :
    key = _hashable ( args ) , _hashable ( kwitems )
except TypeError :
    raise TypeError ( 'Signature is not hashable: args={} kwargs{}' . format ( args , kwargs ) )
return key


signature:
def memoize ( func ) :
    """memoization decorator that respects args and kwargs
 References :
 https : / / wiki . python . org / moin / PythonDecoratorLibrary # Memoize
 Args :
 func ( Callable ) : live python function
 Returns :
 func : memoized wrapper
 CommandLine :
 xdoctest - m ubelt . util _ memoize memoize
 Example :
 > > > import ubelt as ub
 > > > closure = { ' a ' : ' b ' , ' c ' : ' d ' }
 > > > incr = [ 0]
 > > > def foo ( key ) :
 > > > value = closure [ key ]
 > > > incr [ 0 ] + = 1
 > > > return value
 > > > foo _ memo = ub . memoize ( foo )
 > > > assert foo ( ' a ' ) = = ' b ' and foo ( ' c ' ) = = ' d '
 > > > assert incr [ 0 ] = = 2
 > > > print ( ' Call memoized version ' )
 > > > assert foo _ memo ( ' a ' ) = = ' b ' and foo _ memo ( ' c ' ) = = ' d '
 > > > assert incr [ 0 ] = = 4
 > > > assert foo _ memo ( ' a ' ) = = ' b ' and foo _ memo ( ' c ' ) = = ' d '
 > > > print ( ' Counter should no longer increase ' )
 > > > assert incr [ 0 ] = = 4
 > > > print ( ' Closure changes result without memoization ' )
 > > > closure = { ' a ' : 0 , ' c ' : 1}
 > > > assert foo ( ' a ' ) = = 0 and foo ( ' c ' ) = = 1
 > > > assert incr [ 0 ] = = 6
 > > > assert foo _ memo ( ' a ' ) = = ' b ' and foo _ memo ( ' c ' ) = = ' d '"""

implementation:
cache = { }
@ functools . wraps ( func )
def memoizer ( * args , ** kwargs ) :
    key = _make_signature_key ( args , kwargs )
    if key not in cache :
        cache [ key ] = func ( * args , ** kwargs )
    return cache [ key ]
memoizer . cache = cache
return memoizer


signature:
def memoize_property ( fget ) :
    """Return a property attribute for new - style classes that only calls its
 getter on the first access . The result is stored and on subsequent accesses
 is returned , preventing the need to call the getter any more .
 This decorator can either be used by itself or by decorating another
 property . In either case the method will always become a property .
 Notes :
 implementation is a modified version of [ 1 ] .
 References :
 . . [ 1 ] https : / / github . com / estebistec / python - memoized - property
 CommandLine :
 xdoctest - m ubelt . util _ memoize memoize _ property
 Example :
 > > > class C ( object ) :
 . . . load _ name _ count = 0
 . . . @ memoize _ property
 . . . def name ( self ) :
 . . . " name ' s docstring "
 . . . self . load _ name _ count + = 1
 . . . return " the name "
 . . . @ memoize _ property
 . . . @ property
 . . . def another _ name ( self ) :
 . . . " name ' s docstring "
 . . . self . load _ name _ count + = 1
 . . . return " the name "
 > > > c = C ( )
 > > > c . load _ name _ count
 > > > c . name
 ' the name '
 > > > c . load _ name _ count
 > > > c . name
 ' the name '
 > > > c . load _ name _ count
 > > > c . another _ name"""

implementation:
# Unwrap any existing property decorator
while hasattr ( fget , 'fget' ) :
    fget = fget . fget
attr_name = '_' + fget . __name__
@ functools . wraps ( fget )
def fget_memoized ( self ) :
    if not hasattr ( self , attr_name ) :
        setattr ( self , attr_name , fget ( self ) )
    return getattr ( self , attr_name )
return property ( fget_memoized )


signature:
def highlight_code ( text , lexer_name = 'python' , ** kwargs ) :
    """Highlights a block of text using ANSI tags based on language syntax .
 Args :
 text ( str ) : plain text to highlight
 lexer _ name ( str ) : name of language
 * * kwargs : passed to pygments . lexers . get _ lexer _ by _ name
 Returns :
 str : text : highlighted text
 If pygments is not installed , the plain text is returned .
 CommandLine :
 python - c " import pygments . formatters ; print ( list ( pygments . formatters . get _ all _ formatters ( ) ) ) "
 Example :
 > > > import ubelt as ub
 > > > text = ' import ubelt as ub ; print ( ub ) '
 > > > new _ text = ub . highlight _ code ( text )
 > > > print ( new _ text )"""

implementation:
# Resolve extensions to languages
lexer_name = { 'py' : 'python' , 'h' : 'cpp' , 'cpp' : 'cpp' , 'cxx' : 'cpp' , 'c' : 'cpp' , } . get ( lexer_name . replace ( '.' , '' ) , lexer_name )
try :
    import pygments
    import pygments . lexers
    import pygments . formatters
    import pygments . formatters . terminal
    if sys . platform . startswith ( 'win32' ) : # nocover
    # Hack on win32 to support colored output
        import colorama
        colorama . init ( )
    formater = pygments . formatters . terminal . TerminalFormatter ( bg = 'dark' )
    lexer = pygments . lexers . get_lexer_by_name ( lexer_name , ** kwargs )
    new_text = pygments . highlight ( text , lexer , formater )
except ImportError : # nocover
    import warnings
    warnings . warn ( 'pygments is not installed, code will not be highlighted' )
    new_text = text
return new_text


signature:
def color_text ( text , color ) :
    r"""Colorizes text a single color using ansii tags .
 Args :
 text ( str ) : text to colorize
 color ( str ) : may be one of the following : yellow , blink , lightgray ,
 underline , darkyellow , blue , darkblue , faint , fuchsia , black ,
 white , red , brown , turquoise , bold , darkred , darkgreen , reset ,
 standout , darkteal , darkgray , overline , purple , green , teal , fuscia
 Returns :
 str : text : colorized text .
 If pygments is not installed plain text is returned .
 CommandLine :
 python - c " import pygments . console ; print ( sorted ( pygments . console . codes . keys ( ) ) ) "
 python - m ubelt . util _ colors color _ text
 Example :
 > > > text = ' raw text '
 > > > import pytest
 > > > import ubelt as ub
 > > > if ub . modname _ to _ modpath ( ' pygments ' ) :
 > > > # Colors text only if pygments is installed
 > > > assert color _ text ( text , ' red ' ) = = ' \ x1b [ 31;01mraw text \ x1b [ 39;49;00m '
 > > > assert color _ text ( text , None ) = = ' raw text '
 > > > else :
 > > > # Otherwise text passes through unchanged
 > > > assert color _ text ( text , ' red ' ) = = ' raw text '
 > > > assert color _ text ( text , None ) = = ' raw text '"""

implementation:
if color is None :
    return text
try :
    import pygments
    import pygments . console
    if sys . platform . startswith ( 'win32' ) : # nocover
    # Hack on win32 to support colored output
        import colorama
        colorama . init ( )
    ansi_text = pygments . console . colorize ( color , text )
    return ansi_text
except ImportError : # nocover
    import warnings
    warnings . warn ( 'pygments is not installed, text will not be colored' )
    return text


signature:
def iterable ( obj , strok = False ) :
    """Checks if the input implements the iterator interface . An exception is made
 for strings , which return False unless ` strok ` is True
 Args :
 obj ( object ) : a scalar or iterable input
 strok ( bool ) : if True allow strings to be interpreted as iterable
 Returns :
 bool : True if the input is iterable
 Example :
 > > > obj _ list = [ 3 , [ 3 ] , ' 3 ' , ( 3 , ) , [ 3 , 4 , 5 ] , { } ]
 > > > result = [ iterable ( obj ) for obj in obj _ list ]
 > > > assert result = = [ False , True , False , True , True , True ]
 > > > result = [ iterable ( obj , strok = True ) for obj in obj _ list ]
 > > > assert result = = [ False , True , True , True , True , True ]"""

implementation:
try :
    iter ( obj )
except Exception :
    return False
else :
    return strok or not isinstance ( obj , six . string_types )


signature:
def unique ( items , key = None ) :
    """Generates unique items in the order they appear .
 Args :
 items ( Iterable ) : list of items
 key ( Callable , optional ) : custom normalization function .
 If specified returns items where ` key ( item ) ` is unique .
 Yields :
 object : a unique item from the input sequence
 CommandLine :
 python - m utool . util _ list - - exec - unique _ ordered
 Example :
 > > > import ubelt as ub
 > > > items = [ 4 , 6 , 6 , 0 , 6 , 1 , 0 , 2 , 2 , 1]
 > > > unique _ items = list ( ub . unique ( items ) )
 > > > assert unique _ items = = [ 4 , 6 , 0 , 1 , 2]
 Example :
 > > > import ubelt as ub
 > > > items = [ ' A ' , ' a ' , ' b ' , ' B ' , ' C ' , ' c ' , ' D ' , ' e ' , ' D ' , ' E ' ]
 > > > unique _ items = list ( ub . unique ( items , key = six . text _ type . lower ) )
 > > > assert unique _ items = = [ ' A ' , ' b ' , ' C ' , ' D ' , ' e ' ]
 > > > unique _ items = list ( ub . unique ( items ) )
 > > > assert unique _ items = = [ ' A ' , ' a ' , ' b ' , ' B ' , ' C ' , ' c ' , ' D ' , ' e ' , ' E ' ]"""

implementation:
seen = set ( )
if key is None :
    for item in items :
        if item not in seen :
            seen . add ( item )
            yield item
else :
    for item in items :
        norm = key ( item )
        if norm not in seen :
            seen . add ( norm )
            yield item


signature:
def argunique ( items , key = None ) :
    """Returns indices corresponding to the first instance of each unique item .
 Args :
 items ( Sequence ) : indexable collection of items
 key ( Callable , optional ) : custom normalization function .
 If specified returns items where ` key ( item ) ` is unique .
 Yields :
 int : indices of the unique items
 Example :
 > > > items = [ 0 , 2 , 5 , 1 , 1 , 0 , 2 , 4]
 > > > indices = list ( argunique ( items ) )
 > > > assert indices = = [ 0 , 1 , 2 , 3 , 7]
 > > > indices = list ( argunique ( items , key = lambda x : x % 2 = = 0 ) )
 > > > assert indices = = [ 0 , 2]"""

implementation:
# yield from unique ( range ( len ( items ) ) , key = lambda i : items [ i ] )
if key is None :
    return unique ( range ( len ( items ) ) , key = lambda i : items [ i ] )
else :
    return unique ( range ( len ( items ) ) , key = lambda i : key ( items [ i ] ) )


signature:
def unique_flags ( items , key = None ) :
    """Returns a list of booleans corresponding to the first instance of each
 unique item .
 Args :
 items ( Sequence ) : indexable collection of items
 key ( Callable , optional ) : custom normalization function .
 If specified returns items where ` key ( item ) ` is unique .
 Returns :
 List [ bool ] : flags the items that are unique
 Example :
 > > > import ubelt as ub
 > > > items = [ 0 , 2 , 1 , 1 , 0 , 9 , 2]
 > > > flags = unique _ flags ( items )
 > > > assert flags = = [ True , True , True , False , False , True , False ]
 > > > flags = unique _ flags ( items , key = lambda x : x % 2 = = 0)
 > > > assert flags = = [ True , False , True , False , False , False , False ]"""

implementation:
len_ = len ( items )
if key is None :
    item_to_index = dict ( zip ( reversed ( items ) , reversed ( range ( len_ ) ) ) )
    indices = item_to_index . values ( )
else :
    indices = argunique ( items , key = key )
flags = boolmask ( indices , len_ )
return flags


signature:
def boolmask ( indices , maxval = None ) :
    """Constructs a list of booleans where an item is True if its position is in
 ` indices ` otherwise it is False .
 Args :
 indices ( list ) : list of integer indices
 maxval ( int ) : length of the returned list . If not specified
 this is inferred from ` indices `
 Note :
 In the future the arg ` maxval ` may change its name to ` shape `
 Returns :
 list : mask : list of booleans . mask [ idx ] is True if idx in indices
 Example :
 > > > import ubelt as ub
 > > > indices = [ 0 , 1 , 4]
 > > > mask = ub . boolmask ( indices , maxval = 6)
 > > > assert mask = = [ True , True , False , False , True , False ]
 > > > mask = ub . boolmask ( indices )
 > > > assert mask = = [ True , True , False , False , True ]"""

implementation:
if maxval is None :
    indices = list ( indices )
    maxval = max ( indices ) + 1
mask = [ False ] * maxval
for index in indices :
    mask [ index ] = True
return mask


signature:
def allsame ( iterable , eq = operator . eq ) :
    """Determine if all items in a sequence are the same
 Args :
 iterable ( Iterable ) : items to determine if they are all the same
 eq ( Callable , optional ) : function to determine equality
 ( default : operator . eq )
 Example :
 > > > allsame ( [ 1 , 1 , 1 , 1 ] )
 True
 > > > allsame ( [ ] )
 True
 > > > allsame ( [ 0 , 1 ] )
 False
 > > > iterable = iter ( [ 0 , 1 , 1 , 1 ] )
 > > > next ( iterable )
 > > > allsame ( iterable )
 True
 > > > allsame ( range ( 10 ) )
 False
 > > > allsame ( range ( 10 ) , lambda a , b : True )
 True"""

implementation:
iter_ = iter ( iterable )
try :
    first = next ( iter_ )
except StopIteration :
    return True
return all ( eq ( first , item ) for item in iter_ )


signature:
def argsort ( indexable , key = None , reverse = False ) :
    """Returns the indices that would sort a indexable object .
 This is similar to ` numpy . argsort ` , but it is written in pure python and
 works on both lists and dictionaries .
 Args :
 indexable ( Iterable or Mapping ) : indexable to sort by
 key ( Callable , optional ) : customizes the ordering of the indexable
 reverse ( bool , optional ) : if True returns in descending order
 Returns :
 list : indices : list of indices such that sorts the indexable
 Example :
 > > > import ubelt as ub
 > > > # argsort works on dicts by returning keys
 > > > dict _ = { ' a ' : 3 , ' b ' : 2 , ' c ' : 100}
 > > > indices = ub . argsort ( dict _ )
 > > > assert list ( ub . take ( dict _ , indices ) ) = = sorted ( dict _ . values ( ) )
 > > > # argsort works on lists by returning indices
 > > > indexable = [ 100 , 2 , 432 , 10]
 > > > indices = ub . argsort ( indexable )
 > > > assert list ( ub . take ( indexable , indices ) ) = = sorted ( indexable )
 > > > # Can use iterators , but be careful . It exhausts them .
 > > > indexable = reversed ( range ( 100 ) )
 > > > indices = ub . argsort ( indexable )
 > > > assert indices [ 0 ] = = 99
 > > > # Can use key just like sorted
 > > > indexable = [ [ 0 , 1 , 2 ] , [ 3 , 4 ] , [ 5 ] ]
 > > > indices = ub . argsort ( indexable , key = len )
 > > > assert indices = = [ 2 , 1 , 0]
 > > > # Can use reverse just like sorted
 > > > indexable = [ 0 , 2 , 1]
 > > > indices = ub . argsort ( indexable , reverse = True )
 > > > assert indices = = [ 1 , 2 , 0]"""

implementation:
# Create an iterator of value / key pairs
if isinstance ( indexable , collections_abc . Mapping ) :
    vk_iter = ( ( v , k ) for k , v in indexable . items ( ) )
else :
    vk_iter = ( ( v , k ) for k , v in enumerate ( indexable ) )
# Sort by values and extract the indices
if key is None :
    indices = [ k for v , k in sorted ( vk_iter , reverse = reverse ) ]
else : # If key is provided , call it using the value as input
    indices = [ k for v , k in sorted ( vk_iter , key = lambda vk : key ( vk [ 0 ] ) , reverse = reverse ) ]
return indices


signature:
def argmax ( indexable , key = None ) :
    """Returns index / key of the item with the largest value .
 This is similar to ` numpy . argmax ` , but it is written in pure python and
 works on both lists and dictionaries .
 Args :
 indexable ( Iterable or Mapping ) : indexable to sort by
 key ( Callable , optional ) : customizes the ordering of the indexable
 CommandLine :
 python - m ubelt . util _ list argmax
 Example :
 > > > assert argmax ( { ' a ' : 3 , ' b ' : 2 , ' c ' : 100 } ) = = ' c '
 > > > assert argmax ( [ ' a ' , ' c ' , ' b ' , ' z ' , ' f ' ] ) = = 3
 > > > assert argmax ( [ [ 0 , 1 ] , [ 2 , 3 , 4 ] , [ 5 ] ] , key = len ) = = 1
 > > > assert argmax ( { ' a ' : 3 , ' b ' : 2 , 3 : 100 , 4 : 4 } ) = = 3
 > > > assert argmax ( iter ( [ ' a ' , ' c ' , ' b ' , ' z ' , ' f ' ] ) ) = = 3"""

implementation:
if key is None and isinstance ( indexable , collections_abc . Mapping ) :
    return max ( indexable . items ( ) , key = operator . itemgetter ( 1 ) ) [ 0 ]
elif hasattr ( indexable , 'index' ) :
    if key is None :
        return indexable . index ( max ( indexable ) )
    else :
        return indexable . index ( max ( indexable , key = key ) )
else : # less efficient , but catch all solution
    return argsort ( indexable , key = key ) [ - 1 ]


signature:
def argmin ( indexable , key = None ) :
    """Returns index / key of the item with the smallest value .
 This is similar to ` numpy . argmin ` , but it is written in pure python and
 works on both lists and dictionaries .
 Args :
 indexable ( Iterable or Mapping ) : indexable to sort by
 key ( Callable , optional ) : customizes the ordering of the indexable
 Example :
 > > > assert argmin ( { ' a ' : 3 , ' b ' : 2 , ' c ' : 100 } ) = = ' b '
 > > > assert argmin ( [ ' a ' , ' c ' , ' b ' , ' z ' , ' f ' ] ) = = 0
 > > > assert argmin ( [ [ 0 , 1 ] , [ 2 , 3 , 4 ] , [ 5 ] ] , key = len ) = = 2
 > > > assert argmin ( { ' a ' : 3 , ' b ' : 2 , 3 : 100 , 4 : 4 } ) = = ' b '
 > > > assert argmin ( iter ( [ ' a ' , ' c ' , ' A ' , ' z ' , ' f ' ] ) ) = = 2"""

implementation:
if key is None and isinstance ( indexable , collections_abc . Mapping ) :
    return min ( indexable . items ( ) , key = operator . itemgetter ( 1 ) ) [ 0 ]
elif hasattr ( indexable , 'index' ) :
    if key is None :
        return indexable . index ( min ( indexable ) )
    else :
        return indexable . index ( min ( indexable , key = key ) )
else : # less efficient , but catch all solution
    return argsort ( indexable , key = key ) [ 0 ]


signature:
def dzip ( items1 , items2 , cls = dict ) :
    """Zips elementwise pairs between items1 and items2 into a dictionary . Values
 from items2 can be broadcast onto items1.
 Args :
 items1 ( Iterable ) : full sequence
 items2 ( Iterable ) : can either be a sequence of one item or a sequence
 of equal length to ` items1 `
 cls ( Type [ dict ] ) : dictionary type to use . Defaults to dict , but could
 be ordered dict instead .
 Returns :
 dict : similar to dict ( zip ( items1 , items2 ) )
 Example :
 > > > assert dzip ( [ 1 , 2 , 3 ] , [ 4 ] ) = = { 1 : 4 , 2 : 4 , 3 : 4}
 > > > assert dzip ( [ 1 , 2 , 3 ] , [ 4 , 4 , 4 ] ) = = { 1 : 4 , 2 : 4 , 3 : 4}
 > > > assert dzip ( [ ] , [ 4 ] ) = = { }"""

implementation:
try :
    len ( items1 )
except TypeError :
    items1 = list ( items1 )
try :
    len ( items2 )
except TypeError :
    items2 = list ( items2 )
if len ( items1 ) == 0 and len ( items2 ) == 1 : # Corner case :
# allow the first list to be empty and the second list to broadcast a
# value . This means that the equality check wont work for the case
# where items1 and items2 are supposed to correspond , but the length of
# items2 is 1.
    items2 = [ ]
if len ( items2 ) == 1 and len ( items1 ) > 1 :
    items2 = items2 * len ( items1 )
if len ( items1 ) != len ( items2 ) :
    raise ValueError ( 'out of alignment len(items1)=%r, len(items2)=%r' % ( len ( items1 ) , len ( items2 ) ) )
return cls ( zip ( items1 , items2 ) )


signature:
def group_items ( items , groupids ) :
    r"""Groups a list of items by group id .
 Args :
 items ( Iterable ) : a list of items to group
 groupids ( Iterable or Callable ) : a corresponding list of item groupids
 or a function mapping an item to a groupid .
 Returns :
 dict : groupid _ to _ items : maps a groupid to a list of items
 CommandLine :
 python - m ubelt . util _ dict group _ items
 Example :
 > > > import ubelt as ub
 > > > items = [ ' ham ' , ' jam ' , ' spam ' , ' eggs ' , ' cheese ' , ' banana ' ]
 > > > groupids = [ ' protein ' , ' fruit ' , ' protein ' , ' protein ' , ' dairy ' , ' fruit ' ]
 > > > groupid _ to _ items = ub . group _ items ( items , groupids )
 > > > print ( ub . repr2 ( groupid _ to _ items , nl = 0 ) )
 { ' dairy ' : [ ' cheese ' ] , ' fruit ' : [ ' jam ' , ' banana ' ] , ' protein ' : [ ' ham ' , ' spam ' , ' eggs ' ] }"""

implementation:
if callable ( groupids ) :
    keyfunc = groupids
    pair_list = ( ( keyfunc ( item ) , item ) for item in items )
else :
    pair_list = zip ( groupids , items )
# Initialize a dict of lists
groupid_to_items = defaultdict ( list )
# Insert each item into the correct group
for key , item in pair_list :
    groupid_to_items [ key ] . append ( item )
return groupid_to_items


signature:
def dict_hist ( item_list , weight_list = None , ordered = False , labels = None ) :
    """Builds a histogram of items , counting the number of time each item appears
 in the input .
 Args :
 item _ list ( Iterable ) : hashable items ( usually containing duplicates )
 weight _ list ( Iterable ) : corresponding weights for each item
 ordered ( bool ) : if True the result is ordered by frequency
 labels ( Iterable , optional ) : expected labels ( default None )
 Allows this function to pre - initialize the histogram .
 If specified the frequency of each label is initialized to
 zero and item _ list can only contain items specified in labels .
 Returns :
 dict : dictionary where the keys are items in item _ list , and the values
 are the number of times the item appears in item _ list .
 CommandLine :
 python - m ubelt . util _ dict dict _ hist
 Example :
 > > > import ubelt as ub
 > > > item _ list = [ 1 , 2 , 39 , 900 , 1232 , 900 , 1232 , 2 , 2 , 2 , 900]
 > > > hist = ub . dict _ hist ( item _ list )
 > > > print ( ub . repr2 ( hist , nl = 0 ) )
 {1 : 1 , 2 : 4 , 39 : 1 , 900 : 3 , 1232 : 2}
 Example :
 > > > import ubelt as ub
 > > > item _ list = [ 1 , 2 , 39 , 900 , 1232 , 900 , 1232 , 2 , 2 , 2 , 900]
 > > > hist1 = ub . dict _ hist ( item _ list )
 > > > hist2 = ub . dict _ hist ( item _ list , ordered = True )
 > > > try :
 > > > hist3 = ub . dict _ hist ( item _ list , labels = [ ] )
 > > > except KeyError :
 > > > pass
 > > > else :
 > > > raise AssertionError ( ' expected key error ' )
 > > > # result = ub . repr2 ( hist _ )
 > > > weight _ list = [ 1 , 1 , 1 , 0 , 0 , 0 , 0 , 1 , 1 , 1 , 1]
 > > > hist4 = ub . dict _ hist ( item _ list , weight _ list = weight _ list )
 > > > print ( ub . repr2 ( hist1 , nl = 0 ) )
 {1 : 1 , 2 : 4 , 39 : 1 , 900 : 3 , 1232 : 2}
 > > > print ( ub . repr2 ( hist4 , nl = 0 ) )
 {1 : 1 , 2 : 4 , 39 : 1 , 900 : 1 , 1232 : 0}"""

implementation:
if labels is None :
    hist_ = defaultdict ( lambda : 0 )
else :
    hist_ = { k : 0 for k in labels }
if weight_list is None :
    weight_list = it . repeat ( 1 )
# Accumulate frequency
for item , weight in zip ( item_list , weight_list ) :
    hist_ [ item ] += weight
if ordered : # Order by value
    getval = op . itemgetter ( 1 )
    hist = OrderedDict ( [ ( key , value ) for ( key , value ) in sorted ( hist_ . items ( ) , key = getval ) ] )
else : # Cast to a normal dictionary
    hist = dict ( hist_ )
return hist


signature:
def find_duplicates ( items , k = 2 , key = None ) :
    """Find all duplicate items in a list .
 Search for all items that appear more than ` k ` times and return a mapping
 from each ( k ) - duplicate item to the positions it appeared in .
 Args :
 items ( Iterable ) : hashable items possibly containing duplicates
 k ( int ) : only return items that appear at least ` k ` times ( default = 2)
 key ( Callable , optional ) : Returns indices where ` key ( items [ i ] ) `
 maps to a particular value at least k times .
 Returns :
 dict : maps each duplicate item to the indices at which it appears
 CommandLine :
 python - m ubelt . util _ dict find _ duplicates
 Example :
 > > > import ubelt as ub
 > > > items = [ 0 , 0 , 1 , 2 , 3 , 3 , 0 , 12 , 2 , 9]
 > > > duplicates = ub . find _ duplicates ( items )
 > > > print ( ' items = % r ' % ( items , ) )
 > > > print ( ' duplicates = % r ' % ( duplicates , ) )
 > > > assert duplicates = = { 0 : [ 0 , 1 , 6 ] , 2 : [ 3 , 8 ] , 3 : [ 4 , 5 ] }
 > > > assert ub . find _ duplicates ( items , 3 ) = = { 0 : [ 0 , 1 , 6 ] }
 Example :
 > > > import ubelt as ub
 > > > items = [ 0 , 0 , 1 , 2 , 3 , 3 , 0 , 12 , 2 , 9]
 > > > # note : k can be 0
 > > > duplicates = ub . find _ duplicates ( items , k = 0)
 > > > print ( ub . repr2 ( duplicates , nl = 0 ) )
 {0 : [ 0 , 1 , 6 ] , 1 : [ 2 ] , 2 : [ 3 , 8 ] , 3 : [ 4 , 5 ] , 9 : [ 9 ] , 12 : [ 7 ] }
 Example :
 > > > import ubelt as ub
 > > > items = [ 10 , 11 , 12 , 13 , 14 , 15 , 16]
 > > > duplicates = ub . find _ duplicates ( items , key = lambda x : x / / 2)
 > > > print ( ub . repr2 ( duplicates , nl = 0 ) )
 {5 : [ 0 , 1 ] , 6 : [ 2 , 3 ] , 7 : [ 4 , 5 ] }"""

implementation:
# Build mapping from items to the indices at which they appear
# if key is not None :
# items = map ( key , items )
duplicates = defaultdict ( list )
if key is None :
    for count , item in enumerate ( items ) :
        duplicates [ item ] . append ( count )
else :
    for count , item in enumerate ( items ) :
        duplicates [ key ( item ) ] . append ( count )
# remove items seen fewer than k times .
for key in list ( duplicates . keys ( ) ) :
    if len ( duplicates [ key ] ) < k :
        del duplicates [ key ]
duplicates = dict ( duplicates )
return duplicates


signature:
def dict_take ( dict_ , keys , default = util_const . NoParam ) :
    r"""Generates values from a dictionary
 Args :
 dict _ ( Mapping ) : a dictionary to take from
 keys ( Iterable ) : the keys to take
 default ( object , optional ) : if specified uses default if keys are missing
 CommandLine :
 python - m ubelt . util _ dict dict _ take _ gen
 Example :
 > > > import ubelt as ub
 > > > dict _ = { 1 : ' a ' , 2 : ' b ' , 3 : ' c ' }
 > > > keys = [ 1 , 2 , 3 , 4 , 5]
 > > > result = list ( ub . dict _ take ( dict _ , keys , None ) )
 > > > assert result = = [ ' a ' , ' b ' , ' c ' , None , None ]
 Example :
 > > > import ubelt as ub
 > > > dict _ = { 1 : ' a ' , 2 : ' b ' , 3 : ' c ' }
 > > > keys = [ 1 , 2 , 3 , 4 , 5]
 > > > try :
 > > > print ( list ( ub . dict _ take ( dict _ , keys ) ) )
 > > > raise AssertionError ( ' did not get key error ' )
 > > > except KeyError :
 > > > print ( ' correctly got key error ' )"""

implementation:
if default is util_const . NoParam :
    for key in keys :
        yield dict_ [ key ]
else :
    for key in keys :
        yield dict_ . get ( key , default )


signature:
def dict_union ( * args ) :
    """Combines the disjoint keys in multiple dictionaries . For intersecting keys ,
 dictionaries towards the end of the sequence are given precedence .
 Args :
 * args : a sequence of dictionaries
 Returns :
 Dict | OrderedDict :
 OrderedDict if the first argument is an OrderedDict , otherwise dict
 SeeAlso :
 collections . ChainMap - a standard python builtin data structure that
 provides a view that treats multiple dicts as a single dict .
 https : / / docs . python . org / 3 / library / collections . html # chainmap - objects
 Example :
 > > > result = dict _ union ( { ' a ' : 1 , ' b ' : 1 } , { ' b ' : 2 , ' c ' : 2 } )
 > > > assert result = = { ' a ' : 1 , ' b ' : 2 , ' c ' : 2}
 > > > dict _ union ( odict ( [ ( ' a ' , 1 ) , ( ' b ' , 2 ) ] ) , odict ( [ ( ' c ' , 3 ) , ( ' d ' , 4 ) ] ) )
 OrderedDict ( [ ( ' a ' , 1 ) , ( ' b ' , 2 ) , ( ' c ' , 3 ) , ( ' d ' , 4 ) ] )
 > > > dict _ union ( )"""

implementation:
if not args :
    return { }
else :
    dictclass = OrderedDict if isinstance ( args [ 0 ] , OrderedDict ) else dict
    return dictclass ( it . chain . from_iterable ( d . items ( ) for d in args ) )


signature:
def dict_isect ( * args ) :
    """Constructs a dictionary that contains keys common between all inputs .
 The returned values will only belong to the first dictionary .
 Args :
 * args : a sequence of dictionaries ( or sets of keys )
 Returns :
 Dict | OrderedDict :
 OrderedDict if the first argument is an OrderedDict , otherwise dict
 Notes :
 This function can be used as an alternative to ` dict _ subset ` where any
 key not in the dictionary is ignored . See the following example :
 > > > dict _ isect ( { ' a ' : 1 , ' b ' : 2 , ' c ' : 3 } , [ ' a ' , ' c ' , ' d ' ] )
 { ' a ' : 1 , ' c ' : 3}
 Example :
 > > > dict _ isect ( { ' a ' : 1 , ' b ' : 1 } , { ' b ' : 2 , ' c ' : 2 } )
 { ' b ' : 1}
 > > > dict _ isect ( odict ( [ ( ' a ' , 1 ) , ( ' b ' , 2 ) ] ) , odict ( [ ( ' c ' , 3 ) ] ) )
 OrderedDict ( )
 > > > dict _ isect ( )"""

implementation:
if not args :
    return { }
else :
    dictclass = OrderedDict if isinstance ( args [ 0 ] , OrderedDict ) else dict
    common_keys = set . intersection ( * map ( set , args ) )
    first_dict = args [ 0 ]
    return dictclass ( ( k , first_dict [ k ] ) for k in common_keys )


signature:
def map_vals ( func , dict_ ) :
    """applies a function to each of the keys in a dictionary
 Args :
 func ( callable ) : a function or indexable object
 dict _ ( dict ) : a dictionary
 Returns :
 newdict : transformed dictionary
 CommandLine :
 python - m ubelt . util _ dict map _ vals
 Example :
 > > > import ubelt as ub
 > > > dict _ = { ' a ' : [ 1 , 2 , 3 ] , ' b ' : [ ] }
 > > > func = len
 > > > newdict = ub . map _ vals ( func , dict _ )
 > > > assert newdict = = { ' a ' : 3 , ' b ' : 0}
 > > > print ( newdict )
 > > > # Can also use indexables as ` func `
 > > > dict _ = { ' a ' : 0 , ' b ' : 1}
 > > > func = [ 42 , 21]
 > > > newdict = ub . map _ vals ( func , dict _ )
 > > > assert newdict = = { ' a ' : 42 , ' b ' : 21}
 > > > print ( newdict )"""

implementation:
if not hasattr ( func , '__call__' ) :
    func = func . __getitem__
keyval_list = [ ( key , func ( val ) ) for key , val in six . iteritems ( dict_ ) ]
dictclass = OrderedDict if isinstance ( dict_ , OrderedDict ) else dict
newdict = dictclass ( keyval_list )
# newdict = type ( dict _ ) ( keyval _ list )
return newdict


signature:
def invert_dict ( dict_ , unique_vals = True ) :
    r"""Swaps the keys and values in a dictionary .
 Args :
 dict _ ( dict ) : dictionary to invert
 unique _ vals ( bool ) : if False , inverted keys are returned in a set .
 The default is True .
 Returns :
 dict : inverted
 Notes :
 The must values be hashable .
 If the original dictionary contains duplicate values , then only one of
 the corresponding keys will be returned and the others will be
 discarded . This can be prevented by setting ` unique _ vals = True ` ,
 causing the inverted keys to be returned in a set .
 CommandLine :
 python - m ubelt . util _ dict invert _ dict
 Example :
 > > > import ubelt as ub
 > > > dict _ = { ' a ' : 1 , ' b ' : 2}
 > > > inverted = ub . invert _ dict ( dict _ )
 > > > assert inverted = = { 1 : ' a ' , 2 : ' b ' }
 Example :
 > > > import ubelt as ub
 > > > dict _ = ub . odict ( [ ( 2 , ' a ' ) , ( 1 , ' b ' ) , ( 0 , ' c ' ) , ( None , ' d ' ) ] )
 > > > inverted = ub . invert _ dict ( dict _ )
 > > > assert list ( inverted . keys ( ) ) [ 0 ] = = ' a '
 Example :
 > > > import ubelt as ub
 > > > dict _ = { ' a ' : 1 , ' b ' : 0 , ' c ' : 0 , ' d ' : 0 , ' f ' : 2}
 > > > inverted = ub . invert _ dict ( dict _ , unique _ vals = False )
 > > > assert inverted = = { 0 : { ' b ' , ' c ' , ' d ' } , 1 : { ' a ' } , 2 : { ' f ' } }"""

implementation:
if unique_vals :
    if isinstance ( dict_ , OrderedDict ) :
        inverted = OrderedDict ( ( val , key ) for key , val in dict_ . items ( ) )
    else :
        inverted = { val : key for key , val in dict_ . items ( ) }
else : # Handle non - unique keys using groups
    inverted = defaultdict ( set )
    for key , value in dict_ . items ( ) :
        inverted [ value ] . add ( key )
    inverted = dict ( inverted )
return inverted


signature:
def to_dict ( self ) :
    """Recursively casts a AutoDict into a regular dictionary . All nested
 AutoDict values are also converted .
 Returns :
 dict : a copy of this dict without autovivification
 Example :
 > > > from ubelt . util _ dict import AutoDict
 > > > auto = AutoDict ( )
 > > > auto [ 1 ] = 1
 > > > auto [ ' n1 ' ] = AutoDict ( )
 > > > static = auto . to _ dict ( )
 > > > assert not isinstance ( static , AutoDict )
 > > > assert not isinstance ( static [ ' n1 ' ] , AutoDict )"""

implementation: return self . _base ( ( key , ( value . to_dict ( ) if isinstance ( value , AutoDict ) else value ) ) for key , value in self . items ( ) )


signature:
def _win32_can_symlink ( verbose = 0 , force = 0 , testing = 0 ) :
    """CommandLine :
 python - m ubelt . _ win32 _ links _ win32 _ can _ symlink
 Example :
 > > > # xdoc : + REQUIRES ( WIN32)
 > > > import ubelt as ub
 > > > _ win32 _ can _ symlink ( verbose = 1 , force = 1 , testing = 1)"""

implementation:
global __win32_can_symlink__
if verbose :
    print ( '__win32_can_symlink__ = {!r}' . format ( __win32_can_symlink__ ) )
if __win32_can_symlink__ is not None and not force :
    return __win32_can_symlink__
from ubelt import util_platform
tempdir = util_platform . ensure_app_cache_dir ( 'ubelt' , '_win32_can_symlink' )
util_io . delete ( tempdir )
util_path . ensuredir ( tempdir )
dpath = join ( tempdir , 'dpath' )
fpath = join ( tempdir , 'fpath.txt' )
dlink = join ( tempdir , 'dlink' )
flink = join ( tempdir , 'flink.txt' )
util_path . ensuredir ( dpath )
util_io . touch ( fpath )
# Add broken variants of the links for testing purposes
# Its ugly , but so is all this windows code .
if testing :
    broken_dpath = join ( tempdir , 'broken_dpath' )
    broken_fpath = join ( tempdir , 'broken_fpath.txt' )
    # Create files that we will delete after we link to them
    util_path . ensuredir ( broken_dpath )
    util_io . touch ( broken_fpath )
try :
    _win32_symlink ( dpath , dlink )
    if testing :
        _win32_symlink ( broken_dpath , join ( tempdir , 'broken_dlink' ) )
    can_symlink_directories = os . path . islink ( dlink )
except OSError :
    can_symlink_directories = False
if verbose :
    print ( 'can_symlink_directories = {!r}' . format ( can_symlink_directories ) )
try :
    _win32_symlink ( fpath , flink )
    if testing :
        _win32_symlink ( broken_fpath , join ( tempdir , 'broken_flink' ) )
    can_symlink_files = os . path . islink ( flink )
    # os . path . islink ( flink )
except OSError :
    can_symlink_files = False
if verbose :
    print ( 'can_symlink_files = {!r}' . format ( can_symlink_files ) )
assert int ( can_symlink_directories ) + int ( can_symlink_files ) != 1 , ( 'can do one but not both. Unexpected {} {}' . format ( can_symlink_directories , can_symlink_files ) )
try : # test that we can create junctions , even if symlinks are disabled
    djunc = _win32_junction ( dpath , join ( tempdir , 'djunc' ) )
    fjunc = _win32_junction ( fpath , join ( tempdir , 'fjunc.txt' ) )
    if testing :
        _win32_junction ( broken_dpath , join ( tempdir , 'broken_djunc' ) )
        _win32_junction ( broken_fpath , join ( tempdir , 'broken_fjunc.txt' ) )
    assert _win32_is_junction ( djunc )
    assert _win32_is_hardlinked ( fpath , fjunc )
except Exception :
    warnings . warn ( 'We cannot create junctions either!' )
    raise
if testing : # break the links
    util_io . delete ( broken_dpath )
    util_io . delete ( broken_fpath )
    if verbose :
        from ubelt import util_links
        util_links . _dirstats ( tempdir )
try : # Cleanup the test directory
    util_io . delete ( tempdir )
except Exception :
    print ( 'ERROR IN DELETE' )
    from ubelt import util_links
    util_links . _dirstats ( tempdir )
    raise
can_symlink = can_symlink_directories and can_symlink_files
__win32_can_symlink__ = can_symlink
if not can_symlink :
    warnings . warn ( 'Cannot make real symlink. Falling back to junction' )
if verbose :
    print ( 'can_symlink = {!r}' . format ( can_symlink ) )
    print ( '__win32_can_symlink__ = {!r}' . format ( __win32_can_symlink__ ) )
return can_symlink


signature:
def _symlink ( path , link , overwrite = 0 , verbose = 0 ) :
    """Windows helper for ub . symlink"""

implementation:
if exists ( link ) and not os . path . islink ( link ) : # On windows a broken link might still exist as a hard link or a
# junction . Overwrite it if it is a file and we cannot symlink .
# However , if it is a non - junction directory then do not overwrite
    if verbose :
        print ( 'link location already exists' )
    is_junc = _win32_is_junction ( link )
    # NOTE :
    # in python2 broken junctions are directories and exist
    # in python3 broken junctions are directories and do not exist
    if os . path . isdir ( link ) :
        if is_junc :
            pointed = _win32_read_junction ( link )
            if path == pointed :
                if verbose :
                    print ( '...and is a junction that points to the same place' )
                return link
            else :
                if verbose :
                    if not exists ( pointed ) :
                        print ( '...and is a broken junction that points somewhere else' )
                    else :
                        print ( '...and is a junction that points somewhere else' )
        else :
            if verbose :
                print ( '...and is an existing real directory!' )
            raise IOError ( 'Cannot overwrite a real directory' )
    elif os . path . isfile ( link ) :
        if _win32_is_hardlinked ( link , path ) :
            if verbose :
                print ( '...and is a hard link that points to the same place' )
            return link
        else :
            if verbose :
                print ( '...and is a hard link that points somewhere else' )
            if _win32_can_symlink ( ) :
                raise IOError ( 'Cannot overwrite potentially real file if we can symlink' )
    if overwrite :
        if verbose :
            print ( '...overwriting' )
        util_io . delete ( link , verbose > 1 )
    else :
        if exists ( link ) :
            raise IOError ( 'Link already exists' )
_win32_symlink2 ( path , link , verbose = verbose )


signature:
def _win32_symlink2 ( path , link , allow_fallback = True , verbose = 0 ) :
    """Perform a real symbolic link if possible . However , on most versions of
 windows you need special privledges to create a real symlink . Therefore , we
 try to create a symlink , but if that fails we fallback to using a junction .
 AFAIK , the main difference between symlinks and junctions are that symlinks
 can reference relative or absolute paths , where as junctions always
 reference absolute paths . Not 100 % on this though . Windows is weird .
 Note that junctions will not register as links via ` islink ` , but I
 believe real symlinks will ."""

implementation:
if _win32_can_symlink ( ) :
    return _win32_symlink ( path , link , verbose )
else :
    return _win32_junction ( path , link , verbose )


signature:
def _win32_symlink ( path , link , verbose = 0 ) :
    """Creates real symlink . This will only work in versions greater than Windows
 Vista . Creating real symlinks requires admin permissions or at least
 specially enabled symlink permissions . On Windows 10 enabling developer
 mode should give you these permissions ."""

implementation:
from ubelt import util_cmd
if os . path . isdir ( path ) : # directory symbolic link
    if verbose :
        print ( '... as directory symlink' )
    command = 'mklink /D "{}" "{}"' . format ( link , path )
    # Using the win32 API seems to result in privilege errors
    # but using shell commands does not have this problem . Weird .
    # jwfs . symlink ( path , link , target _ is _ directory = True )
    # TODO : what do we need to do to use the windows api instead of shell ?
else : # file symbolic link
    if verbose :
        print ( '... as file symlink' )
    command = 'mklink "{}" "{}"' . format ( link , path )
if command is not None :
    info = util_cmd . cmd ( command , shell = True )
    if info [ 'ret' ] != 0 :
        from ubelt import util_format
        permission_msg = 'You do not have sufficient privledges'
        if permission_msg not in info [ 'err' ] :
            print ( 'Failed command:' )
            print ( info [ 'command' ] )
            print ( util_format . repr2 ( info , nl = 1 ) )
        raise OSError ( str ( info ) )
return link


signature:
def _win32_junction ( path , link , verbose = 0 ) :
    """On older ( pre 10 ) versions of windows we need admin privledges to make
 symlinks , however junctions seem to work .
 For paths we do a junction ( softlink ) and for files we use a hard link
 CommandLine :
 python - m ubelt . _ win32 _ links _ win32 _ junction
 Example :
 > > > # xdoc : + REQUIRES ( WIN32)
 > > > import ubelt as ub
 > > > root = ub . ensure _ app _ cache _ dir ( ' ubelt ' , ' win32 _ junction ' )
 > > > ub . delete ( root )
 > > > ub . ensuredir ( root )
 > > > fpath = join ( root , ' fpath . txt ' )
 > > > dpath = join ( root , ' dpath ' )
 > > > fjunc = join ( root , ' fjunc . txt ' )
 > > > djunc = join ( root , ' djunc ' )
 > > > ub . touch ( fpath )
 > > > ub . ensuredir ( dpath )
 > > > ub . ensuredir ( join ( root , ' djunc _ fake ' ) )
 > > > ub . ensuredir ( join ( root , ' djunc _ fake with space ' ) )
 > > > ub . touch ( join ( root , ' djunc _ fake with space file ' ) )
 > > > _ win32 _ junction ( fpath , fjunc )
 > > > _ win32 _ junction ( dpath , djunc )
 > > > # thank god colons are not allowed
 > > > djunc2 = join ( root , ' djunc2 [ with pathological attrs ] ' )
 > > > _ win32 _ junction ( dpath , djunc2)
 > > > _ win32 _ is _ junction ( djunc )
 > > > ub . writeto ( join ( djunc , ' afile . txt ' ) , ' foo ' )
 > > > assert ub . readfrom ( join ( dpath , ' afile . txt ' ) ) = = ' foo '
 > > > ub . writeto ( fjunc , ' foo ' )"""

implementation:
# junctions store absolute paths
path = os . path . abspath ( path )
link = os . path . abspath ( link )
from ubelt import util_cmd
if os . path . isdir ( path ) : # try using a junction ( soft link )
    if verbose :
        print ( '... as soft link' )
    # TODO : what is the windows api for this ?
    command = 'mklink /J "{}" "{}"' . format ( link , path )
else : # try using a hard link
    if verbose :
        print ( '... as hard link' )
    # command = ' mklink / H " { } " " { } " ' . format ( link , path )
    try :
        jwfs . link ( path , link )
        # this seems to be allowed
    except Exception :
        print ( 'Failed to hardlink link={} to path={}' . format ( link , path ) )
        raise
    command = None
if command is not None :
    info = util_cmd . cmd ( command , shell = True )
    if info [ 'ret' ] != 0 :
        from ubelt import util_format
        print ( 'Failed command:' )
        print ( info [ 'command' ] )
        print ( util_format . repr2 ( info , nl = 1 ) )
        raise OSError ( str ( info ) )
return link


signature:
def _win32_is_junction ( path ) :
    """Determines if a path is a win32 junction
 CommandLine :
 python - m ubelt . _ win32 _ links _ win32 _ is _ junction
 Example :
 > > > # xdoc : + REQUIRES ( WIN32)
 > > > import ubelt as ub
 > > > root = ub . ensure _ app _ cache _ dir ( ' ubelt ' , ' win32 _ junction ' )
 > > > ub . delete ( root )
 > > > ub . ensuredir ( root )
 > > > dpath = join ( root , ' dpath ' )
 > > > djunc = join ( root , ' djunc ' )
 > > > ub . ensuredir ( dpath )
 > > > _ win32 _ junction ( dpath , djunc )
 > > > assert _ win32 _ is _ junction ( djunc ) is True
 > > > assert _ win32 _ is _ junction ( dpath ) is False
 > > > assert _ win32 _ is _ junction ( ' notafile ' ) is False"""

implementation:
if not exists ( path ) :
    if os . path . isdir ( path ) :
        if not os . path . islink ( path ) :
            return True
    return False
return jwfs . is_reparse_point ( path ) and not os . path . islink ( path )


signature:
def _win32_read_junction ( path ) :
    """Returns the location that the junction points , raises ValueError if path is
 not a junction .
 CommandLine :
 python - m ubelt . _ win32 _ links _ win32 _ read _ junction
 Example :
 > > > # xdoc : + REQUIRES ( WIN32)
 > > > import ubelt as ub
 > > > root = ub . ensure _ app _ cache _ dir ( ' ubelt ' , ' win32 _ junction ' )
 > > > ub . delete ( root )
 > > > ub . ensuredir ( root )
 > > > dpath = join ( root , ' dpath ' )
 > > > djunc = join ( root , ' djunc ' )
 > > > ub . ensuredir ( dpath )
 > > > _ win32 _ junction ( dpath , djunc )
 > > > path = djunc
 > > > pointed = _ win32 _ read _ junction ( path )
 > > > print ( ' pointed = { ! r } ' . format ( pointed ) )"""

implementation:
if not jwfs . is_reparse_point ( path ) :
    raise ValueError ( 'not a junction' )
# - - - Older version based on using shell commands - - -
# if not exists ( path ) :
# if six . PY2:
# raise OSError ( ' Cannot find path = { } ' . format ( path ) )
# else :
# raise FileNotFoundError ( ' Cannot find path = { } ' . format ( path ) )
# target _ name = os . path . basename ( path )
# for type _ or _ size , name , pointed in _ win32 _ dir ( path , ' * ' ) :
# if type _ or _ size = = ' < JUNCTION > ' and name = = target _ name :
# return pointed
# raise ValueError ( ' not a junction ' )
# new version using the windows api
handle = jwfs . api . CreateFile ( path , 0 , 0 , None , jwfs . api . OPEN_EXISTING , jwfs . api . FILE_FLAG_OPEN_REPARSE_POINT | jwfs . api . FILE_FLAG_BACKUP_SEMANTICS , None )
if handle == jwfs . api . INVALID_HANDLE_VALUE :
    raise WindowsError ( )
res = jwfs . reparse . DeviceIoControl ( handle , jwfs . api . FSCTL_GET_REPARSE_POINT , None , 10240 )
bytes = jwfs . create_string_buffer ( res )
p_rdb = jwfs . cast ( bytes , jwfs . POINTER ( jwfs . api . REPARSE_DATA_BUFFER ) )
rdb = p_rdb . contents
if rdb . tag not in [ 2684354563 , jwfs . api . IO_REPARSE_TAG_SYMLINK ] :
    raise RuntimeError ( "Expected <2684354563 or 2684354572>, but got %d" % rdb . tag )
jwfs . handle_nonzero_success ( jwfs . api . CloseHandle ( handle ) )
subname = rdb . get_substitute_name ( )
# probably has something to do with long paths , not sure
if subname . startswith ( '?\\' ) :
    subname = subname [ 2 : ]
return subname


signature:
def _win32_rmtree ( path , verbose = 0 ) :
    """rmtree for win32 that treats junctions like directory symlinks .
 The junction removal portion may not be safe on race conditions .
 There is a known issue that prevents shutil . rmtree from
 deleting directories with junctions .
 https : / / bugs . python . org / issue31226"""

implementation:
# - - - old version using the shell - - -
# def _ rmjunctions ( root ) :
# subdirs = [ ]
# for type _ or _ size , name , pointed in _ win32 _ dir ( root ) :
# if type _ or _ size = = ' < DIR > ' :
# subdirs . append ( name )
# elif type _ or _ size = = ' < JUNCTION > ' :
# # remove any junctions as we encounter them
# # os . unlink ( join ( root , name ) )
# os . rmdir ( join ( root , name ) )
# # recurse in all real directories
# for name in subdirs :
# _ rmjunctions ( join ( root , name ) )
def _rmjunctions ( root ) :
    subdirs = [ ]
    for name in os . listdir ( root ) :
        current = join ( root , name )
        if os . path . isdir ( current ) :
            if _win32_is_junction ( current ) : # remove any junctions as we encounter them
                os . rmdir ( current )
            elif not os . path . islink ( current ) :
                subdirs . append ( current )
    # recurse in all real directories
    for subdir in subdirs :
        _rmjunctions ( subdir )
if _win32_is_junction ( path ) :
    if verbose :
        print ( 'Deleting <JUNCTION> directory="{}"' . format ( path ) )
    os . rmdir ( path )
else :
    if verbose :
        print ( 'Deleting directory="{}"' . format ( path ) )
    # first remove all junctions
    _rmjunctions ( path )
    # now we can rmtree as normal
    import shutil
    shutil . rmtree ( path )


signature:
def _win32_is_hardlinked ( fpath1 , fpath2 ) :
    """Test if two hard links point to the same location
 CommandLine :
 python - m ubelt . _ win32 _ links _ win32 _ is _ hardlinked
 Example :
 > > > # xdoc : + REQUIRES ( WIN32)
 > > > import ubelt as ub
 > > > root = ub . ensure _ app _ cache _ dir ( ' ubelt ' , ' win32 _ hardlink ' )
 > > > ub . delete ( root )
 > > > ub . ensuredir ( root )
 > > > fpath1 = join ( root , ' fpath1 ' )
 > > > fpath2 = join ( root , ' fpath2 ' )
 > > > ub . touch ( fpath1)
 > > > ub . touch ( fpath2)
 > > > fjunc1 = _ win32 _ junction ( fpath1 , join ( root , ' fjunc1 ' ) )
 > > > fjunc2 = _ win32 _ junction ( fpath2 , join ( root , ' fjunc2 ' ) )
 > > > assert _ win32 _ is _ hardlinked ( fjunc1 , fpath1)
 > > > assert _ win32 _ is _ hardlinked ( fjunc2 , fpath2)
 > > > assert not _ win32 _ is _ hardlinked ( fjunc2 , fpath1)
 > > > assert not _ win32 _ is _ hardlinked ( fjunc1 , fpath2)"""

implementation:
# NOTE : jwf . samefile ( fpath1 , fpath2 ) seems to behave differently
def get_read_handle ( fpath ) :
    if os . path . isdir ( fpath ) :
        dwFlagsAndAttributes = jwfs . api . FILE_FLAG_BACKUP_SEMANTICS
    else :
        dwFlagsAndAttributes = 0
    hFile = jwfs . api . CreateFile ( fpath , jwfs . api . GENERIC_READ , jwfs . api . FILE_SHARE_READ , None , jwfs . api . OPEN_EXISTING , dwFlagsAndAttributes , None )
    return hFile
def get_unique_id ( hFile ) :
    info = jwfs . api . BY_HANDLE_FILE_INFORMATION ( )
    res = jwfs . api . GetFileInformationByHandle ( hFile , info )
    jwfs . handle_nonzero_success ( res )
    unique_id = ( info . volume_serial_number , info . file_index_high , info . file_index_low )
    return unique_id
hFile1 = get_read_handle ( fpath1 )
hFile2 = get_read_handle ( fpath2 )
try :
    are_equal = ( get_unique_id ( hFile1 ) == get_unique_id ( hFile2 ) )
except Exception :
    raise
finally :
    jwfs . api . CloseHandle ( hFile1 )
    jwfs . api . CloseHandle ( hFile2 )
return are_equal


signature:
def _win32_dir ( path , star = '' ) :
    """Using the windows cmd shell to get information about a directory"""

implementation:
from ubelt import util_cmd
import re
wrapper = 'cmd /S /C "{}"'
# the / S will preserve all inner quotes
command = 'dir /-C "{}"{}' . format ( path , star )
wrapped = wrapper . format ( command )
info = util_cmd . cmd ( wrapped , shell = True )
if info [ 'ret' ] != 0 :
    from ubelt import util_format
    print ( 'Failed command:' )
    print ( info [ 'command' ] )
    print ( util_format . repr2 ( info , nl = 1 ) )
    raise OSError ( str ( info ) )
# parse the output of dir to get some info
# Remove header and footer
lines = info [ 'out' ] . split ( '\n' ) [ 5 : - 3 ]
splitter = re . compile ( '( +)' )
for line in lines :
    parts = splitter . split ( line )
    date , sep , time , sep , ampm , sep , type_or_size , sep = parts [ : 8 ]
    name = '' . join ( parts [ 8 : ] )
    # if type is a junction then name will also contain the linked loc
    if name == '.' or name == '..' :
        continue
    if type_or_size in [ '<JUNCTION>' , '<SYMLINKD>' , '<SYMLINK>' ] : # colons cannot be in path names , so use that to find where
    # the name ends
        pos = name . find ( ':' )
        bpos = name [ : pos ] . rfind ( '[' )
        name = name [ : bpos - 1 ]
        pointed = name [ bpos + 1 : - 1 ]
        yield type_or_size , name , pointed
    else :
        yield type_or_size , name , None


signature:
def parse_generator_doubling ( config ) :
    """Returns generators that double with each value returned
 Config includes optional start value"""

implementation:
start = 1
if 'start' in config :
    start = int ( config [ 'start' ] )
# We cannot simply use start as the variable , because of scoping
# limitations
def generator ( ) :
    val = start
    while ( True ) :
        yield val
        val = val * 2
return generator ( )


signature:
def parse ( config ) :
    """Parse a contains validator , which takes as the config a simple string to find"""

implementation:
if not isinstance ( config , basestring ) :
    raise TypeError ( "Contains input must be a simple string" )
validator = ContainsValidator ( )
validator . contains_string = config
return validator


signature:
def retrieve_adjacency_matrix ( graph , order_nodes = None , weight = False ) :
    """Retrieve the adjacency matrix from the nx . DiGraph or numpy array ."""

implementation:
if isinstance ( graph , np . ndarray ) :
    return graph
elif isinstance ( graph , nx . DiGraph ) :
    if order_nodes is None :
        order_nodes = graph . nodes ( )
    if not weight :
        return np . array ( nx . adjacency_matrix ( graph , order_nodes , weight = None ) . todense ( ) )
    else :
        return np . array ( nx . adjacency_matrix ( graph , order_nodes ) . todense ( ) )
else :
    raise TypeError ( "Only networkx.DiGraph and np.ndarray (adjacency matrixes) are supported." )


signature:
def precision_recall ( target , prediction , low_confidence_undirected = False ) :
    r"""Compute precision - recall statistics for directed graphs .
 Precision recall statistics are useful to compare algorithms that make
 predictions with a confidence score . Using these statistics , performance
 of an algorithms given a set threshold ( confidence score ) can be
 approximated .
 Area under the precision - recall curve , as well as the coordinates of the
 precision recall curve are computed , using the scikit - learn library tools .
 Note that unlike the AUROC metric , this metric does not account for class
 imbalance .
 Precision is defined by : : math : ` Pr = tp / ( tp + fp ) ` and directly denotes the
 total classification accuracy given a confidence threshold . On the other
 hand , Recall is defined by : : math : ` Re = tp / ( tp + fn ) ` and denotes
 misclassification given a threshold .
 Args :
 target ( numpy . ndarray or networkx . DiGraph ) : Target graph , must be of
 ones and zeros .
 prediction ( numpy . ndarray or networkx . DiGraph ) : Prediction made by the
 algorithm to evaluate .
 low _ confidence _ undirected : Put the lowest confidence possible to
 undirected edges ( edges that are symmetric in the confidence score ) .
 Default : False
 Returns :
 tuple : tuple containing :
 + Area under the precision recall curve ( float )
 + Tuple of data points of the precision - recall curve used in the computation of the score ( tuple ) .
 Examples :
 > > > import numpy as np
 > > > tar , pred = np . random . randint ( 2 , size = ( 10 , 10 ) ) , np . random . randn ( 10 , 10)
 > > > # adjacency matrixes of size 10x10
 > > > aupr , curve = precision _ recall ( target , input )
 > > > # leave low _ confidence _ undirected to False as the predictions are continuous"""

implementation:
true_labels = retrieve_adjacency_matrix ( target )
pred = retrieve_adjacency_matrix ( prediction , target . nodes ( ) if isinstance ( target , nx . DiGraph ) else None , weight = True )
if low_confidence_undirected : # Take account of undirected edges by putting them with low confidence
    pred [ pred == pred . transpose ( ) ] *= min ( min ( pred [ np . nonzero ( pred ) ] ) * .5 , .1 )
precision , recall , _ = precision_recall_curve ( true_labels . ravel ( ) , pred . ravel ( ) )
aupr = auc ( recall , precision , reorder = True )
return aupr , list ( zip ( precision , recall ) )


signature:
def SHD ( target , pred , double_for_anticausal = True ) :
    r"""Compute the Structural Hamming Distance .
 The Structural Hamming Distance ( SHD ) is a standard distance to compare
 graphs by their adjacency matrix . It consists in computing the difference
 between the two ( binary ) adjacency matrixes : every edge that is either
 missing or not in the target graph is counted as a mistake . Note that
 for directed graph , two mistakes can be counted as the edge in the wrong
 direction is false and the edge in the good direction is missing ; the
 ` double _ for _ anticausal ` argument accounts for this remark . Setting it to
 ` False ` will count this as a single mistake .
 Args :
 target ( numpy . ndarray or networkx . DiGraph ) : Target graph , must be of
 ones and zeros .
 prediction ( numpy . ndarray or networkx . DiGraph ) : Prediction made by the
 algorithm to evaluate .
 double _ for _ anticausal ( bool ) : Count the badly oriented edges as two
 mistakes . Default : True
 Returns :
 int : Structural Hamming Distance ( int ) .
 The value tends to zero as the graphs tend to be identical .
 Examples :
 > > > from numpy . random import randint
 > > > tar , pred = randint ( 2 , size = ( 10 , 10 ) ) , randint ( 2 , size = ( 10 , 10 ) )
 > > > SHD ( tar , pred , double _ for _ anticausal = False )"""

implementation:
true_labels = retrieve_adjacency_matrix ( target )
predictions = retrieve_adjacency_matrix ( pred , target . nodes ( ) if isinstance ( target , nx . DiGraph ) else None )
diff = np . abs ( true_labels - predictions )
if double_for_anticausal :
    return np . sum ( diff )
else :
    diff = diff + diff . transpose ( )
    diff [ diff > 1 ] = 1
    # Ignoring the double edges .
    return np . sum ( diff ) / 2


signature:
def SID ( target , pred ) :
    """Compute the Strutural Intervention Distance .
 [ R wrapper ] The Structural Intervention Distance ( SID ) is a new distance
 for graphs introduced by Peters and Bühlmann ( 2013 ) . This distance was
 created to account for the shortcomings of the SHD metric for a causal
 sense .
 It consists in computing the path between all the pairs of variables , and
 checks if the causal relationship between the variables is respected .
 The given graphs have to be DAGs for the SID metric to make sense .
 Args :
 target ( numpy . ndarray or networkx . DiGraph ) : Target graph , must be of
 ones and zeros , and instance of either numpy . ndarray or
 networkx . DiGraph . Must be a DAG .
 prediction ( numpy . ndarray or networkx . DiGraph ) : Prediction made by the
 algorithm to evaluate . Must be a DAG .
 Returns :
 int : Structural Intervention Distance .
 The value tends to zero as the graphs tends to be identical .
 . . note : :
 Ref : Structural Intervention Distance ( SID ) for Evaluating Causal Graphs ,
 Jonas Peters , Peter Bühlmann : https : / / arxiv . org / abs / 1306.1043
 Examples :
 > > > from numpy . random import randint
 > > > tar = np . triu ( randint ( 2 , size = ( 10 , 10 ) ) )
 > > > pred = np . triu ( randint ( 2 , size = ( 10 , 10 ) ) )
 > > > SID ( tar , pred )"""

implementation:
if not RPackages . SID :
    raise ImportError ( "SID R package is not available. Please check your installation." )
true_labels = retrieve_adjacency_matrix ( target )
predictions = retrieve_adjacency_matrix ( pred , target . nodes ( ) if isinstance ( target , nx . DiGraph ) else None )
os . makedirs ( '/tmp/cdt_SID/' )
def retrieve_result ( ) :
    return np . loadtxt ( '/tmp/cdt_SID/result.csv' )
try :
    np . savetxt ( '/tmp/cdt_SID/target.csv' , true_labels , delimiter = ',' )
    np . savetxt ( '/tmp/cdt_SID/pred.csv' , predictions , delimiter = ',' )
    sid_score = launch_R_script ( "{}/R_templates/sid.R" . format ( os . path . dirname ( os . path . realpath ( __file__ ) ) ) , { "{target}" : '/tmp/cdt_SID/target.csv' , "{prediction}" : '/tmp/cdt_SID/pred.csv' , "{result}" : '/tmp/cdt_SID/result.csv' } , output_function = retrieve_result )
# Cleanup
except Exception as e :
    rmtree ( '/tmp/cdt_SID' )
    raise e
except KeyboardInterrupt :
    rmtree ( '/tmp/cdt_SID/' )
    raise KeyboardInterrupt
rmtree ( '/tmp/cdt_SID' )
return sid_score


signature:
def create_graph_from_data ( self , data , ** kwargs ) :
    """Apply causal discovery on observational data using CCDr .
 Args :
 data ( pandas . DataFrame ) : DataFrame containing the data
 Returns :
 networkx . DiGraph : Solution given by the CCDR algorithm ."""

implementation:
# Building setup w / arguments .
self . arguments [ '{VERBOSE}' ] = str ( self . verbose ) . upper ( )
results = self . _run_ccdr ( data , verbose = self . verbose )
return nx . relabel_nodes ( nx . DiGraph ( results ) , { idx : i for idx , i in enumerate ( data . columns ) } )


signature:
def init_variables ( self , verbose = False ) :
    """Redefine the causes of the graph ."""

implementation:
for j in range ( 1 , self . nodes ) :
    nb_parents = np . random . randint ( 0 , min ( [ self . parents_max , j ] ) + 1 )
    for i in np . random . choice ( range ( 0 , j ) , nb_parents , replace = False ) :
        self . adjacency_matrix [ i , j ] = 1
try :
    self . g = nx . DiGraph ( self . adjacency_matrix )
    assert not list ( nx . simple_cycles ( self . g ) )
except AssertionError :
    if verbose :
        print ( "Regenerating, graph non valid..." )
    self . init_variables ( )
# Mechanisms
self . cfunctions = [ self . mechanism ( int ( sum ( self . adjacency_matrix [ : , i ] ) ) , self . points , self . noise , noise_coeff = self . noise_coeff ) if sum ( self . adjacency_matrix [ : , i ] ) else self . initial_generator for i in range ( self . nodes ) ]


signature:
def generate ( self , rescale = True ) :
    """Generate data from an FCM containing cycles ."""

implementation:
if self . cfunctions is None :
    self . init_variables ( )
for i in nx . topological_sort ( self . g ) : # Root cause
    if not sum ( self . adjacency_matrix [ : , i ] ) :
        self . data [ 'V{}' . format ( i ) ] = self . cfunctions [ i ] ( self . points )
    # Generating causes
    else :
        self . data [ 'V{}' . format ( i ) ] = self . cfunctions [ i ] ( self . data . iloc [ : , self . adjacency_matrix [ : , i ] . nonzero ( ) [ 0 ] ] . values )
    if rescale :
        self . data [ 'V{}' . format ( i ) ] = scale ( self . data [ 'V{}' . format ( i ) ] . values )
return self . g , self . data


signature:
def to_csv ( self , fname_radical , ** kwargs ) :
    """Save data to the csv format by default , in two separate files .
 Optional keyword arguments can be passed to pandas ."""

implementation:
if self . data is not None :
    self . data . to_csv ( fname_radical + '_data.csv' , index = False , ** kwargs )
    pd . DataFrame ( self . adjacency_matrix ) . to_csv ( fname_radical + '_target.csv' , index = False , ** kwargs )
else :
    raise ValueError ( "Graph has not yet been generated. \
                              Use self.generate() to do so." )


signature:
def launch_R_script ( template , arguments , output_function = None , verbose = True , debug = False ) :
    """Launch an R script , starting from a template and replacing text in file
 before execution .
 Args :
 template ( str ) : path to the template of the R script
 arguments ( dict ) : Arguments that modify the template ' s placeholders
 with arguments
 output _ function ( function ) : Function to execute * * after * * the execution
 of the R script , and its output is returned by this function . Used
 traditionally as a function to retrieve the results of the
 execution .
 verbose ( bool ) : Sets the verbosity of the R subprocess .
 debug ( bool ) : If True , the generated scripts are not deleted .
 Return :
 Returns the output of the ` ` output _ function ` ` if not ` None `
 else ` True ` or ` False ` depending on whether the execution was
 successful ."""

implementation:
id = str ( uuid . uuid4 ( ) )
os . makedirs ( '/tmp/cdt_R_script_' + id + '/' )
try :
    scriptpath = '/tmp/cdt_R_script_' + id + '/instance_{}' . format ( os . path . basename ( template ) )
    copy ( template , scriptpath )
    with fileinput . FileInput ( scriptpath , inplace = True ) as file :
        for line in file :
            mline = line
            for elt in arguments :
                mline = mline . replace ( elt , arguments [ elt ] )
            print ( mline , end = '' )
    if output_function is None :
        output = subprocess . call ( "Rscript --vanilla {}" . format ( scriptpath ) , shell = True , stdout = subprocess . DEVNULL , stderr = subprocess . DEVNULL )
    else :
        if verbose :
            process = subprocess . Popen ( "Rscript --vanilla {}" . format ( scriptpath ) , shell = True )
        else :
            process = subprocess . Popen ( "Rscript --vanilla {}" . format ( scriptpath ) , shell = True , stdout = subprocess . DEVNULL , stderr = subprocess . DEVNULL )
        process . wait ( )
        output = output_function ( )
# Cleaning up
except Exception as e :
    if not debug :
        rmtree ( '/tmp/cdt_R_script_' + id + '/' )
    raise e
except KeyboardInterrupt :
    if not debug :
        rmtree ( '/tmp/cdt_R_script_' + id + '/' )
    raise KeyboardInterrupt
if not debug :
    rmtree ( '/tmp/cdt_R_script_' + id + '/' )
return output


signature:
def check_R_package ( self , package ) :
    """Execute a subprocess to check the package ' s availability .
 Args :
 package ( str ) : Name of the package to be tested .
 Returns :
 bool : ` True ` if the package is available , ` False ` otherwise"""

implementation:
test_package = not bool ( launch_R_script ( "{}/R_templates/test_import.R" . format ( os . path . dirname ( os . path . realpath ( __file__ ) ) ) , { "{package}" : package } , verbose = True ) )
return test_package


signature:
def bin_variable ( var , bins = 'fd' ) : # bin with normalization
    """Bin variables w / normalization ."""

implementation:
var = np . array ( var ) . astype ( np . float )
var = ( var - np . mean ( var ) ) / np . std ( var )
var = np . digitize ( var , np . histogram ( var , bins = bins ) [ 1 ] )
return var


signature:
def predict ( self , a , b , ** kwargs ) :
    """Perform the independence test .
 : param a : input data
 : param b : input data
 : type a : array - like , numerical data
 : type b : array - like , numerical data
 : return : dependency statistic ( 1 = Highly dependent , 0 = Not dependent )
 : rtype : float"""

implementation:
binning_alg = kwargs . get ( 'bins' , 'fd' )
return metrics . adjusted_mutual_info_score ( bin_variable ( a , bins = binning_alg ) , bin_variable ( b , bins = binning_alg ) )


signature:
def predict ( self , df_data , graph = None , ** kwargs ) :
    """Orient a graph using the method defined by the arguments .
 Depending on the type of ` graph ` , this function process to execute
 different functions :
 1 . If ` ` graph ` ` is a ` ` networkx . DiGraph ` ` , then ` ` self . orient _ directed _ graph ` ` is executed .
 2 . If ` ` graph ` ` is a ` ` networkx . Graph ` ` , then ` ` self . orient _ undirected _ graph ` ` is executed .
 3 . If ` ` graph ` ` is a ` ` None ` ` , then ` ` self . create _ graph _ from _ data ` ` is executed .
 Args :
 df _ data ( pandas . DataFrame ) : DataFrame containing the observational data .
 graph ( networkx . DiGraph or networkx . Graph or None ) : Prior knowledge on the causal graph .
 . . warning : :
 Requirement : Name of the nodes in the graph must correspond to the
 name of the variables in df _ data"""

implementation:
if graph is None :
    return self . create_graph_from_data ( df_data , ** kwargs )
elif isinstance ( graph , nx . DiGraph ) :
    return self . orient_directed_graph ( df_data , graph , ** kwargs )
elif isinstance ( graph , nx . Graph ) :
    return self . orient_undirected_graph ( df_data , graph , ** kwargs )
else :
    print ( 'Unknown Graph type' )
    raise ValueError


signature:
def graph_evaluation ( data , adj_matrix , gpu = None , gpu_id = 0 , ** kwargs ) :
    """Evaluate a graph taking account of the hardware ."""

implementation:
gpu = SETTINGS . get_default ( gpu = gpu )
device = 'cuda:{}' . format ( gpu_id ) if gpu else 'cpu'
obs = th . FloatTensor ( data ) . to ( device )
cgnn = CGNN_model ( adj_matrix , data . shape [ 0 ] , gpu_id = gpu_id , ** kwargs ) . to ( device )
cgnn . reset_parameters ( )
return cgnn . run ( obs , ** kwargs )


signature:
def parallel_graph_evaluation ( data , adj_matrix , nb_runs = 16 , nb_jobs = None , ** kwargs ) :
    """Parallelize the various runs of CGNN to evaluate a graph ."""

implementation:
nb_jobs = SETTINGS . get_default ( nb_jobs = nb_jobs )
if nb_runs == 1 :
    return graph_evaluation ( data , adj_matrix , ** kwargs )
else :
    output = Parallel ( n_jobs = nb_jobs ) ( delayed ( graph_evaluation ) ( data , adj_matrix , idx = run , gpu_id = run % SETTINGS . GPU , ** kwargs ) for run in range ( nb_runs ) )
    return np . mean ( output )


signature:
def hill_climbing ( data , graph , ** kwargs ) :
    """Hill Climbing optimization : a greedy exploration algorithm ."""

implementation:
nodelist = list ( data . columns )
data = scale ( data . values ) . astype ( 'float32' )
tested_candidates = [ nx . adj_matrix ( graph , nodelist = nodelist , weight = None ) ]
best_score = parallel_graph_evaluation ( data , tested_candidates [ 0 ] . todense ( ) , ** kwargs )
best_candidate = graph
can_improve = True
while can_improve :
    can_improve = False
    for ( i , j ) in best_candidate . edges ( ) :
        test_graph = deepcopy ( best_candidate )
        test_graph . add_edge ( j , i , weight = test_graph [ i ] [ j ] [ 'weight' ] )
        test_graph . remove_edge ( i , j )
        tadjmat = nx . adj_matrix ( test_graph , nodelist = nodelist , weight = None )
        if ( nx . is_directed_acyclic_graph ( test_graph ) and not any ( [ ( tadjmat != cand ) . nnz == 0 for cand in tested_candidates ] ) ) :
            tested_candidates . append ( tadjmat )
            score = parallel_graph_evaluation ( data , tadjmat . todense ( ) , ** kwargs )
            if score < best_score :
                can_improve = True
                best_candidate = test_graph
                best_score = score
                break
return best_candidate


signature:
def forward ( self ) :
    """Generate according to the topological order of the graph ."""

implementation:
self . noise . data . normal_ ( )
if not self . confounding :
    for i in self . topological_order :
        self . generated [ i ] = self . blocks [ i ] ( th . cat ( [ v for c in [ [ self . generated [ j ] for j in np . nonzero ( self . adjacency_matrix [ : , i ] ) [ 0 ] ] , [ self . noise [ : , [ i ] ] ] ] for v in c ] , 1 ) )
else :
    for i in self . topological_order :
        self . generated [ i ] = self . blocks [ i ] ( th . cat ( [ v for c in [ [ self . generated [ j ] for j in np . nonzero ( self . adjacency_matrix [ : , i ] ) [ 0 ] ] , [ self . corr_noise [ min ( i , j ) , max ( i , j ) ] for j in np . nonzero ( self . i_adj_matrix [ : , i ] ) [ 0 ] ] [ self . noise [ : , [ i ] ] ] ] for v in c ] , 1 ) )
return th . cat ( self . generated , 1 )


signature:
def run ( self , data , train_epochs = 1000 , test_epochs = 1000 , verbose = None , idx = 0 , lr = 0.01 , ** kwargs ) :
    """Run the CGNN on a given graph ."""

implementation:
verbose = SETTINGS . get_default ( verbose = verbose )
optim = th . optim . Adam ( self . parameters ( ) , lr = lr )
self . score . zero_ ( )
with trange ( train_epochs + test_epochs , disable = not verbose ) as t :
    for epoch in t :
        optim . zero_grad ( )
        generated_data = self . forward ( )
        mmd = self . criterion ( generated_data , data )
        if not epoch % 200 :
            t . set_postfix ( idx = idx , epoch = epoch , loss = mmd . item ( ) )
        mmd . backward ( )
        optim . step ( )
        if epoch >= test_epochs :
            self . score . add_ ( mmd . data )
return self . score . cpu ( ) . numpy ( ) / test_epochs


signature:
def create_graph_from_data ( self , data ) :
    """Use CGNN to create a graph from scratch . All the possible structures
 are tested , which leads to a super exponential complexity . It would be
 preferable to start from a graph skeleton for large graphs .
 Args :
 data ( pandas . DataFrame ) : Observational data on which causal
 discovery has to be performed .
 Returns :
 networkx . DiGraph : Solution given by CGNN ."""

implementation:
warnings . warn ( "An exhaustive search of the causal structure of CGNN without" " skeleton is super-exponential in the number of variables." )
# Building all possible candidates :
nb_vars = len ( list ( data . columns ) )
data = scale ( data . values ) . astype ( 'float32' )
candidates = [ np . reshape ( np . array ( i ) , ( nb_vars , nb_vars ) ) for i in itertools . product ( [ 0 , 1 ] , repeat = nb_vars * nb_vars ) if ( np . trace ( np . reshape ( np . array ( i ) , ( nb_vars , nb_vars ) ) ) == 0 and nx . is_directed_acyclic_graph ( nx . DiGraph ( np . reshape ( np . array ( i ) , ( nb_vars , nb_vars ) ) ) ) ) ]
warnings . warn ( "A total of {} graphs will be evaluated." . format ( len ( candidates ) ) )
scores = [ parallel_graph_evaluation ( data , i , nh = self . nh , nb_runs = self . nb_runs , gpu = self . gpu , nb_jobs = self . nb_jobs , lr = self . lr , train_epochs = self . train_epochs , test_epochs = self . test_epochs , verbose = self . verbose ) for i in candidates ]
final_candidate = candidates [ scores . index ( min ( scores ) ) ]
output = np . zeros ( final_candidate . shape )
# Retrieve the confidence score on each edge .
for ( i , j ) , x in np . ndenumerate ( final_candidate ) :
    if x > 0 :
        cand = final_candidate
        cand [ i , j ] = 0
        output [ i , j ] = min ( scores ) - scores [ candidates . index ( cand ) ]
return nx . DiGraph ( candidates [ output ] , { idx : i for idx , i in enumerate ( data . columns ) } )


signature:
def orient_directed_graph ( self , data , dag , alg = 'HC' ) :
    """Modify and improve a directed acyclic graph solution using CGNN .
 Args :
 data ( pandas . DataFrame ) : Observational data on which causal
 discovery has to be performed .
 dag ( nx . DiGraph ) : Graph that provides the initial solution ,
 on which the CGNN algorithm will be applied .
 alg ( str ) : Exploration heuristic to use , among [ " HC " , " HCr " ,
 " tabu " , " EHC " ]
 Returns :
 networkx . DiGraph : Solution given by CGNN ."""

implementation:
alg_dic = { 'HC' : hill_climbing , 'HCr' : hill_climbing_with_removal , 'tabu' : tabu_search , 'EHC' : exploratory_hill_climbing }
return alg_dic [ alg ] ( data , dag , nh = self . nh , nb_runs = self . nb_runs , gpu = self . gpu , nb_jobs = self . nb_jobs , lr = self . lr , train_epochs = self . train_epochs , test_epochs = self . test_epochs , verbose = self . verbose )


signature:
def orient_undirected_graph ( self , data , umg , alg = 'HC' ) :
    """Orient the undirected graph using GNN and apply CGNN to improve the graph .
 Args :
 data ( pandas . DataFrame ) : Observational data on which causal
 discovery has to be performed .
 umg ( nx . Graph ) : Graph that provides the skeleton , on which the GNN
 then the CGNN algorithm will be applied .
 alg ( str ) : Exploration heuristic to use , among [ " HC " , " HCr " ,
 " tabu " , " EHC " ]
 Returns :
 networkx . DiGraph : Solution given by CGNN .
 . . note : :
 GNN ( ` ` cdt . causality . pairwise . GNN ` ` ) is first used to orient the
 undirected graph and output a DAG before applying CGNN ."""

implementation:
warnings . warn ( "The pairwise GNN model is computed on each edge of the UMG " "to initialize the model and start CGNN with a DAG" )
gnn = GNN ( nh = self . nh , lr = self . lr )
og = gnn . orient_graph ( data , umg , nb_runs = self . nb_runs , nb_max_runs = self . nb_runs , nb_jobs = self . nb_jobs , train_epochs = self . train_epochs , test_epochs = self . test_epochs , verbose = self . verbose , gpu = self . gpu )
# Pairwise method
# print ( nx . adj _ matrix ( og ) . todense ( ) . shape )
# print ( list ( og . edges ( ) ) )
dag = dagify_min_edge ( og )
# print ( nx . adj _ matrix ( dag ) . todense ( ) . shape )
return self . orient_directed_graph ( data , dag , alg = alg )


signature:
def eval_entropy ( x ) :
    """Evaluate the entropy of the input variable .
 : param x : input variable 1D
 : return : entropy of x"""

implementation:
hx = 0.
sx = sorted ( x )
for i , j in zip ( sx [ : - 1 ] , sx [ 1 : ] ) :
    delta = j - i
    if bool ( delta ) :
        hx += np . log ( np . abs ( delta ) )
hx = hx / ( len ( x ) - 1 ) + psi ( len ( x ) ) - psi ( 1 )
return hx


signature:
def integral_approx_estimator ( x , y ) :
    """Integral approximation estimator for causal inference .
 : param x : input variable x 1D
 : param y : input variable y 1D
 : return : Return value of the IGCI model > 0 if x - > y otherwise if return < 0"""


signature:
def predict_proba ( self , a , b , ** kwargs ) :
    """Evaluate a pair using the IGCI model .
 : param a : Input variable 1D
 : param b : Input variable 1D
 : param kwargs : { refMeasure : Scaling method ( gaussian , integral or None ) ,
 estimator : method used to evaluate the pairs ( entropy or integral ) }
 : return : Return value of the IGCI model > 0 if a - > b otherwise if return < 0"""

implementation:
estimators = { 'entropy' : lambda x , y : eval_entropy ( y ) - eval_entropy ( x ) , 'integral' : integral_approx_estimator }
ref_measures = { 'gaussian' : lambda x : standard_scale . fit_transform ( x . reshape ( ( - 1 , 1 ) ) ) , 'uniform' : lambda x : min_max_scale . fit_transform ( x . reshape ( ( - 1 , 1 ) ) ) , 'None' : lambda x : x }
ref_measure = ref_measures [ kwargs . get ( 'refMeasure' , 'gaussian' ) ]
estimator = estimators [ kwargs . get ( 'estimator' , 'entropy' ) ]
a = ref_measure ( a )
b = ref_measure ( b )
return estimator ( a , b )


signature:
def featurize_row ( self , x , y ) :
    """Projects the causal pair to the RKHS using the sampled kernel approximation .
 Args :
 x ( np . ndarray ) : Variable 1
 y ( np . ndarray ) : Variable 2
 Returns :
 np . ndarray : projected empirical distributions into a single fixed - size vector ."""

implementation:
x = x . ravel ( )
y = y . ravel ( )
b = np . ones ( x . shape )
dx = np . cos ( np . dot ( self . W2 , np . vstack ( ( x , b ) ) ) ) . mean ( 1 )
dy = np . cos ( np . dot ( self . W2 , np . vstack ( ( y , b ) ) ) ) . mean ( 1 )
if ( sum ( dx ) > sum ( dy ) ) :
    return np . hstack ( ( dx , dy , np . cos ( np . dot ( self . W , np . vstack ( ( x , y , b ) ) ) ) . mean ( 1 ) ) )
else :
    return np . hstack ( ( dx , dy , np . cos ( np . dot ( self . W , np . vstack ( ( y , x , b ) ) ) ) . mean ( 1 ) ) )


signature:
def fit ( self , x , y ) :
    """Train the model .
 Args :
 x _ tr ( pd . DataFrame ) : CEPC format dataframe containing the pairs
 y _ tr ( pd . DataFrame or np . ndarray ) : labels associated to the pairs"""

implementation:
train = np . vstack ( ( np . array ( [ self . featurize_row ( row . iloc [ 0 ] , row . iloc [ 1 ] ) for idx , row in x . iterrows ( ) ] ) , np . array ( [ self . featurize_row ( row . iloc [ 1 ] , row . iloc [ 0 ] ) for idx , row in x . iterrows ( ) ] ) ) )
labels = np . vstack ( ( y , - y ) ) . ravel ( )
verbose = 1 if self . verbose else 0
self . clf = CLF ( verbose = verbose , min_samples_leaf = self . L , n_estimators = self . E , max_depth = self . max_depth , n_jobs = self . n_jobs ) . fit ( train , labels )


signature:
def predict_proba ( self , x , y = None , ** kwargs ) :
    """Predict the causal score using a trained RCC model
 Args :
 x ( numpy . array or pandas . DataFrame or pandas . Series ) : First variable or dataset .
 args ( numpy . array ) : second variable ( optional depending on the 1st argument ) .
 Returns :
 float : Causation score ( Value : 1 if a - > b and - 1 if b - > a )"""

implementation:
if self . clf is None :
    raise ValueError ( "Model has to be trained before making predictions." )
if x is pandas . Series :
    input_ = self . featurize_row ( x . iloc [ 0 ] , x . iloc [ 1 ] ) . reshape ( ( 1 , - 1 ) )
elif x is pandas . DataFrame :
    input_ = np . array ( [ self . featurize_row ( x . iloc [ 0 ] , x . iloc [ 1 ] ) for row in x ] )
elif y is not None :
    input_ = self . featurize_row ( x , y ) . reshape ( ( 1 , - 1 ) )
else :
    raise TypeError ( "DataType not understood." )
return self . clf . predict ( input_ )


signature:
def predict_features ( self , df_features , df_target , nh = 20 , idx = 0 , dropout = 0. , activation_function = th . nn . ReLU , lr = 0.01 , l1 = 0.1 , batch_size = - 1 , train_epochs = 1000 , test_epochs = 1000 , device = None , verbose = None , nb_runs = 3 ) :
    """For one variable , predict its neighbours .
 Args :
 df _ features ( pandas . DataFrame ) :
 df _ target ( pandas . Series ) :
 nh ( int ) : number of hidden units
 idx ( int ) : ( optional ) for printing purposes
 dropout ( float ) : probability of dropout ( between 0 and 1)
 activation _ function ( torch . nn . Module ) : activation function of the NN
 lr ( float ) : learning rate of Adam
 l1 ( float ) : L1 penalization coefficient
 batch _ size ( int ) : batch size , defaults to full - batch
 train _ epochs ( int ) : number of train epochs
 test _ epochs ( int ) : number of test epochs
 device ( str ) : cuda or cpu device ( defaults to ` ` cdt . SETTINGS . default _ device ` ` )
 verbose ( bool ) : verbosity ( defaults to ` ` cdt . SETTINGS . verbose ` ` )
 nb _ runs ( int ) : number of bootstrap runs
 Returns :
 list : scores of each feature relatively to the target"""

implementation:
device , verbose = SETTINGS . get_default ( ( 'device' , device ) , ( 'verbose' , verbose ) )
x = th . FloatTensor ( scale ( df_features . values ) ) . to ( device )
y = th . FloatTensor ( scale ( df_target . values ) ) . to ( device )
out = [ ]
for i in range ( nb_runs ) :
    model = FSGNN_model ( [ x . size ( ) [ 1 ] + 1 , nh , 1 ] , dropout = dropout , activation_function = activation_function ) . to ( device )
    out . append ( model . train ( x , y , lr = 0.01 , l1 = 0.1 , batch_size = - 1 , train_epochs = train_epochs , test_epochs = test_epochs , device = device , verbose = verbose ) )
return list ( np . mean ( np . array ( out ) , axis = 0 ) )


signature:
def predict_undirected_graph ( self , data ) :
    """Build a skeleton using a pairwise independence criterion .
 Args :
 data ( pandas . DataFrame ) : Raw data table
 Returns :
 networkx . Graph : Undirected graph representing the skeleton ."""

implementation:
graph = Graph ( )
for idx_i , i in enumerate ( data . columns ) :
    for idx_j , j in enumerate ( data . columns [ idx_i + 1 : ] ) :
        score = self . predict ( data [ i ] . values , data [ j ] . values )
        if abs ( score ) > 0.001 :
            graph . add_edge ( i , j , weight = score )
return graph


signature:
def run_feature_selection ( self , df_data , target , idx = 0 , ** kwargs ) :
    """Run feature selection for one node : wrapper around
 ` ` self . predict _ features ` ` .
 Args :
 df _ data ( pandas . DataFrame ) : All the observational data
 target ( str ) : Name of the target variable
 idx ( int ) : ( optional ) For printing purposes
 Returns :
 list : scores of each feature relatively to the target"""

implementation:
list_features = list ( df_data . columns . values )
list_features . remove ( target )
df_target = pd . DataFrame ( df_data [ target ] , columns = [ target ] )
df_features = df_data [ list_features ]
return self . predict_features ( df_features , df_target , idx = idx , ** kwargs )


signature:
def predict ( self , df_data , threshold = 0.05 , ** kwargs ) :
    """Predict the skeleton of the graph from raw data .
 Returns iteratively the feature selection algorithm on each node .
 Args :
 df _ data ( pandas . DataFrame ) : data to construct a graph from
 threshold ( float ) : cutoff value for feature selection scores
 kwargs ( dict ) : additional arguments for algorithms
 Returns :
 networkx . Graph : predicted skeleton of the graph ."""

implementation:
nb_jobs = kwargs . get ( "nb_jobs" , SETTINGS . NB_JOBS )
list_nodes = list ( df_data . columns . values )
if nb_jobs != 1 :
    result_feature_selection = Parallel ( n_jobs = nb_jobs ) ( delayed ( self . run_feature_selection ) ( df_data , node , idx , ** kwargs ) for idx , node in enumerate ( list_nodes ) )
else :
    result_feature_selection = [ self . run_feature_selection ( df_data , node , idx , ** kwargs ) for idx , node in enumerate ( list_nodes ) ]
for idx , i in enumerate ( result_feature_selection ) :
    try :
        i . insert ( idx , 0 )
    except AttributeError : # if results are numpy arrays
        result_feature_selection [ idx ] = np . insert ( i , idx , 0 )
matrix_results = np . array ( result_feature_selection )
matrix_results *= matrix_results . transpose ( )
np . fill_diagonal ( matrix_results , 0 )
matrix_results /= 2
graph = nx . Graph ( )
for ( i , j ) , x in np . ndenumerate ( matrix_results ) :
    if matrix_results [ i , j ] > threshold :
        graph . add_edge ( list_nodes [ i ] , list_nodes [ j ] , weight = matrix_results [ i , j ] )
for node in list_nodes :
    if node not in graph . nodes ( ) :
        graph . add_node ( node )
return graph


signature:
def orient_undirected_graph ( self , data , graph ) :
    """Run GIES on an undirected graph .
 Args :
 data ( pandas . DataFrame ) : DataFrame containing the data
 graph ( networkx . Graph ) : Skeleton of the graph to orient
 Returns :
 networkx . DiGraph : Solution given by the GIES algorithm ."""

implementation:
# Building setup w / arguments .
self . arguments [ '{VERBOSE}' ] = str ( self . verbose ) . upper ( )
self . arguments [ '{SCORE}' ] = self . scores [ self . score ]
fe = DataFrame ( nx . adj_matrix ( graph , weight = None ) . todense ( ) )
fg = DataFrame ( 1 - fe . values )
results = self . _run_gies ( data , fixedGaps = fg , verbose = self . verbose )
return nx . relabel_nodes ( nx . DiGraph ( results ) , { idx : i for idx , i in enumerate ( data . columns ) } )


signature:
def create_graph_from_data ( self , data ) :
    """Run the GIES algorithm .
 Args :
 data ( pandas . DataFrame ) : DataFrame containing the data
 Returns :
 networkx . DiGraph : Solution given by the GIES algorithm ."""

implementation:
# Building setup w / arguments .
self . arguments [ '{SCORE}' ] = self . scores [ self . score ]
self . arguments [ '{VERBOSE}' ] = str ( self . verbose ) . upper ( )
results = self . _run_gies ( data , verbose = self . verbose )
return nx . relabel_nodes ( nx . DiGraph ( results ) , { idx : i for idx , i in enumerate ( data . columns ) } )


signature:
def _run_gies ( self , data , fixedGaps = None , verbose = True ) :
    """Setting up and running GIES with all arguments ."""

implementation:
# Run gies
id = str ( uuid . uuid4 ( ) )
os . makedirs ( '/tmp/cdt_gies' + id + '/' )
self . arguments [ '{FOLDER}' ] = '/tmp/cdt_gies' + id + '/'
def retrieve_result ( ) :
    return read_csv ( '/tmp/cdt_gies' + id + '/result.csv' , delimiter = ',' ) . values
try :
    data . to_csv ( '/tmp/cdt_gies' + id + '/data.csv' , header = False , index = False )
    if fixedGaps is not None :
        fixedGaps . to_csv ( '/tmp/cdt_gies' + id + '/fixedgaps.csv' , index = False , header = False )
        self . arguments [ '{SKELETON}' ] = 'TRUE'
    else :
        self . arguments [ '{SKELETON}' ] = 'FALSE'
    gies_result = launch_R_script ( "{}/R_templates/gies.R" . format ( os . path . dirname ( os . path . realpath ( __file__ ) ) ) , self . arguments , output_function = retrieve_result , verbose = verbose )
# Cleanup
except Exception as e :
    rmtree ( '/tmp/cdt_gies' + id + '' )
    raise e
except KeyboardInterrupt :
    rmtree ( '/tmp/cdt_gies' + id + '/' )
    raise KeyboardInterrupt
rmtree ( '/tmp/cdt_gies' + id + '' )
return gies_result


signature:
def plot_curves ( i_batch , adv_loss , gen_loss , l1_reg , cols ) :
    """Plot SAM ' s various losses ."""

implementation:
from matplotlib import pyplot as plt
if i_batch == 0 :
    try :
        ax . clear ( )
        ax . plot ( range ( len ( adv_plt ) ) , adv_plt , "r-" , linewidth = 1.5 , markersize = 4 , label = "Discriminator" )
        ax . plot ( range ( len ( adv_plt ) ) , gen_plt , "g-" , linewidth = 1.5 , markersize = 4 , label = "Generators" )
        ax . plot ( range ( len ( adv_plt ) ) , l1_plt , "b-" , linewidth = 1.5 , markersize = 4 , label = "L1-Regularization" )
        plt . legend ( )
        adv_plt . append ( adv_loss . cpu ( ) . data [ 0 ] )
        gen_plt . append ( gen_loss . cpu ( ) . data [ 0 ] / cols )
        l1_plt . append ( l1_reg . cpu ( ) . data [ 0 ] )
        plt . pause ( 0.0001 )
    except NameError :
        plt . ion ( )
        fig , ax = plt . figure ( )
        plt . xlabel ( "Epoch" )
        plt . ylabel ( "Losses" )
        plt . pause ( 0.0001 )
        adv_plt = [ adv_loss . cpu ( ) . data [ 0 ] ]
        gen_plt = [ gen_loss . cpu ( ) . data [ 0 ] / cols ]
        l1_plt = [ l1_reg . cpu ( ) . data [ 0 ] ]
else :
    adv_plt . append ( adv_loss . cpu ( ) . data [ 0 ] )
    gen_plt . append ( gen_loss . cpu ( ) . data [ 0 ] / cols )
    l1_plt . append ( l1_reg . cpu ( ) . data [ 0 ] )


signature:
def plot_gen ( epoch , batch , generated_variables , pairs_to_plot = [ [ 0 , 1 ] ] ) :
    """Plot generated pairs of variables ."""

implementation:
from matplotlib import pyplot as plt
if epoch == 0 :
    plt . ion ( )
plt . clf ( )
for ( i , j ) in pairs_to_plot :
    plt . scatter ( generated_variables [ i ] . data . cpu ( ) . numpy ( ) , batch . data . cpu ( ) . numpy ( ) [ : , j ] , label = "Y -> X" )
    plt . scatter ( batch . data . cpu ( ) . numpy ( ) [ : , i ] , generated_variables [ j ] . data . cpu ( ) . numpy ( ) , label = "X -> Y" )
    plt . scatter ( batch . data . cpu ( ) . numpy ( ) [ : , i ] , batch . data . cpu ( ) . numpy ( ) [ : , j ] , label = "original data" )
    plt . legend ( )
plt . pause ( 0.01 )


signature:
def run_SAM ( df_data , skeleton = None , ** kwargs ) :
    """Execute the SAM model .
 : param df _ data : Input data ; either np . array or pd . DataFrame"""

implementation:
gpu = kwargs . get ( 'gpu' , False )
gpu_no = kwargs . get ( 'gpu_no' , 0 )
train_epochs = kwargs . get ( 'train_epochs' , 1000 )
test_epochs = kwargs . get ( 'test_epochs' , 1000 )
batch_size = kwargs . get ( 'batch_size' , - 1 )
lr_gen = kwargs . get ( 'lr_gen' , 0.1 )
lr_disc = kwargs . get ( 'lr_disc' , lr_gen )
verbose = kwargs . get ( 'verbose' , True )
regul_param = kwargs . get ( 'regul_param' , 0.1 )
dnh = kwargs . get ( 'dnh' , None )
plot = kwargs . get ( "plot" , False )
plot_generated_pair = kwargs . get ( "plot_generated_pair" , False )
d_str = "Epoch: {} -- Disc: {} -- Gen: {} -- L1: {}"
try :
    list_nodes = list ( df_data . columns )
    df_data = ( df_data [ list_nodes ] ) . values
except AttributeError :
    list_nodes = list ( range ( df_data . shape [ 1 ] ) )
data = df_data . astype ( 'float32' )
data = th . from_numpy ( data )
if batch_size == - 1 :
    batch_size = data . shape [ 0 ]
rows , cols = data . size ( )
# Get the list of indexes to ignore
if skeleton is not None :
    zero_components = [ [ ] for i in range ( cols ) ]
    skel = nx . adj_matrix ( skeleton , weight = None )
    for i , j in zip ( * ( ( 1 - skel ) . nonzero ( ) ) ) :
        zero_components [ j ] . append ( i )
else :
    zero_components = [ [ i ] for i in range ( cols ) ]
sam = SAM_generators ( ( rows , cols ) , zero_components , batch_norm = True , ** kwargs )
# Begin UGLY
activation_function = kwargs . get ( 'activation_function' , th . nn . Tanh )
try :
    del kwargs [ "activation_function" ]
except KeyError :
    pass
discriminator_sam = SAM_discriminator ( [ cols , dnh , dnh , 1 ] , batch_norm = True , activation_function = th . nn . LeakyReLU , activation_argument = 0.2 , ** kwargs )
kwargs [ "activation_function" ] = activation_function
# End of UGLY
if gpu :
    sam = sam . cuda ( gpu_no )
    discriminator_sam = discriminator_sam . cuda ( gpu_no )
    data = data . cuda ( gpu_no )
# Select parameters to optimize : ignore the non connected nodes
criterion = th . nn . BCEWithLogitsLoss ( )
g_optimizer = th . optim . Adam ( sam . parameters ( ) , lr = lr_gen )
d_optimizer = th . optim . Adam ( discriminator_sam . parameters ( ) , lr = lr_disc )
true_variable = Variable ( th . ones ( batch_size , 1 ) , requires_grad = False )
false_variable = Variable ( th . zeros ( batch_size , 1 ) , requires_grad = False )
causal_filters = th . zeros ( data . shape [ 1 ] , data . shape [ 1 ] )
if gpu :
    true_variable = true_variable . cuda ( gpu_no )
    false_variable = false_variable . cuda ( gpu_no )
    causal_filters = causal_filters . cuda ( gpu_no )
data_iterator = DataLoader ( data , batch_size = batch_size , shuffle = True )
# TRAIN
for epoch in range ( train_epochs + test_epochs ) :
    for i_batch , batch in enumerate ( data_iterator ) :
        batch = Variable ( batch )
        batch_vectors = [ batch [ : , [ i ] ] for i in range ( cols ) ]
        g_optimizer . zero_grad ( )
        d_optimizer . zero_grad ( )
        # Train the discriminator
        generated_variables = sam ( batch )
        disc_losses = [ ]
        gen_losses = [ ]
        for i in range ( cols ) :
            generator_output = th . cat ( [ v for c in [ batch_vectors [ : i ] , [ generated_variables [ i ] ] , batch_vectors [ i + 1 : ] ] for v in c ] , 1 )
            # 1 . Train discriminator on fake
            disc_output_detached = discriminator_sam ( generator_output . detach ( ) )
            disc_output = discriminator_sam ( generator_output )
            disc_losses . append ( criterion ( disc_output_detached , false_variable ) )
            # 2 . Train the generator :
            gen_losses . append ( criterion ( disc_output , true_variable ) )
        true_output = discriminator_sam ( batch )
        adv_loss = sum ( disc_losses ) / cols + criterion ( true_output , true_variable )
        gen_loss = sum ( gen_losses )
        adv_loss . backward ( )
        d_optimizer . step ( )
        # 3 . Compute filter regularization
        filters = th . stack ( [ i . fs_filter [ 0 , : - 1 ] . abs ( ) for i in sam . blocks ] , 1 )
        l1_reg = regul_param * filters . sum ( )
        loss = gen_loss + l1_reg
        if verbose and epoch % 200 == 0 and i_batch == 0 :
            print ( str ( i ) + " " + d_str . format ( epoch , adv_loss . item ( ) , gen_loss . item ( ) / cols , l1_reg . item ( ) ) )
        loss . backward ( )
        # STORE ASSYMETRY values for output
        if epoch > train_epochs :
            causal_filters . add_ ( filters . data )
        g_optimizer . step ( )
        if plot :
            plot_curves ( i_batch , adv_loss , gen_loss , l1_reg , cols )
        if plot_generated_pair and epoch % 200 == 0 :
            plot_gen ( epoch , batch , generated_variables )
return causal_filters . div_ ( test_epochs ) . cpu ( ) . numpy ( )


signature:
def reset_parameters ( self ) :
    """Reset the parameters ."""

implementation:
stdv = 1. / math . sqrt ( self . weight . size ( 1 ) )
self . weight . data . uniform_ ( - stdv , stdv )
if self . bias is not None :
    self . bias . data . uniform_ ( - stdv , stdv )


signature:
def forward ( self , input ) :
    """Feed - forward through the network ."""

implementation: return th . nn . functional . linear ( input , self . weight . div ( self . weight . pow ( 2 ) . sum ( 0 ) . sqrt ( ) ) )


signature:
def forward ( self , x ) :
    """Feed - forward the model ."""

implementation: return self . layers ( x * ( self . _filter * self . fs_filter ) . expand_as ( x ) )


implementation:
for i in self . noise :
    i . data . normal_ ( )
self . generated_variables = [ self . blocks [ i ] ( th . cat ( [ x , self . noise [ i ] ] , 1 ) ) for i in range ( self . cols ) ]
return self . generated_variables


signature:
def predict ( self , data , graph = None , nruns = 6 , njobs = None , gpus = 0 , verbose = None , plot = False , plot_generated_pair = False , return_list_results = False ) :
    """Execute SAM on a dataset given a skeleton or not .
 Args :
 data ( pandas . DataFrame ) : Observational data for estimation of causal relationships by SAM
 skeleton ( numpy . ndarray ) : A priori knowledge about the causal relationships as an adjacency matrix .
 Can be fed either directed or undirected links .
 nruns ( int ) : Number of runs to be made for causal estimation .
 Recommended : > = 12 for optimal performance .
 njobs ( int ) : Numbers of jobs to be run in Parallel .
 Recommended : 1 if no GPU available , 2 * number of GPUs else .
 gpus ( int ) : Number of available GPUs for the algorithm .
 verbose ( bool ) : verbose mode
 plot ( bool ) : Plot losses interactively . Not recommended if nruns > 1
 plot _ generated _ pair ( bool ) : plots a generated pair interactively . Not recommended if nruns > 1
 Returns :
 networkx . DiGraph : Graph estimated by SAM , where A [ i , j ] is the term
 of the ith variable for the jth generator ."""

implementation:
verbose , njobs = SETTINGS . get_default ( ( 'verbose' , verbose ) , ( 'nb_jobs' , njobs ) )
if njobs != 1 :
    list_out = Parallel ( n_jobs = njobs ) ( delayed ( run_SAM ) ( data , skeleton = graph , lr_gen = self . lr , lr_disc = self . dlr , regul_param = self . l1 , nh = self . nh , dnh = self . dnh , gpu = bool ( gpus ) , train_epochs = self . train , test_epochs = self . test , batch_size = self . batchsize , plot = plot , verbose = verbose , gpu_no = idx % max ( gpus , 1 ) ) for idx in range ( nruns ) )
else :
    list_out = [ run_SAM ( data , skeleton = graph , lr_gen = self . lr , lr_disc = self . dlr , regul_param = self . l1 , nh = self . nh , dnh = self . dnh , gpu = bool ( gpus ) , train_epochs = self . train , test_epochs = self . test , batch_size = self . batchsize , plot = plot , verbose = verbose , gpu_no = 0 ) for idx in range ( nruns ) ]
if return_list_results :
    return list_out
else :
    W = list_out [ 0 ]
    for w in list_out [ 1 : ] :
        W += w
    W /= nruns
return nx . relabel_nodes ( nx . DiGraph ( W ) , { idx : i for idx , i in enumerate ( data . columns ) } )


signature:
def predict_proba ( self , a , b , ** kwargs ) :
    """Infer causal relationships between 2 variables using the RECI statistic
 : param a : Input variable 1
 : param b : Input variable 2
 : return : Causation coefficient ( Value : 1 if a - > b and - 1 if b - > a )
 : rtype : float"""

implementation: return self . b_fit_score ( b , a ) - self . b_fit_score ( a , b )


signature:
def b_fit_score ( self , x , y ) :
    """Compute the RECI fit score
 Args :
 x ( numpy . ndarray ) : Variable 1
 y ( numpy . ndarray ) : Variable 2
 Returns :
 float : RECI fit score"""

implementation:
x = np . reshape ( minmax_scale ( x ) , ( - 1 , 1 ) )
y = np . reshape ( minmax_scale ( y ) , ( - 1 , 1 ) )
poly = PolynomialFeatures ( degree = self . degree )
poly_x = poly . fit_transform ( x )
poly_x [ : , 1 ] = 0
poly_x [ : , 2 ] = 0
regressor = LinearRegression ( )
regressor . fit ( poly_x , y )
y_predict = regressor . predict ( poly_x )
error = mean_squared_error ( y_predict , y )
return error


signature:
def predict_proba ( self , a , b , ** kwargs ) :
    """Infer causal relationships between 2 variables using the CDS statistic
 Args :
 a ( numpy . ndarray ) : Variable 1
 b ( numpy . ndarray ) : Variable 2
 Returns :
 float : Causation score ( Value : 1 if a - > b and - 1 if b - > a )"""

implementation: return self . cds_score ( b , a ) - self . cds_score ( a , b )