Haskell --> Python Tools

htool.py

"""     htool   :: Haskell Tool

Module of borrowed operators from Haskell implemented in Python for the purpose
of gaining more familiarity with functional programming in a familiar
environment. 
"""
__author__  = 'CjHanks'
__license__ = 'Free' 

from operator  import lt, gt 
from sys       import version_info 
from itertools import chain, combinations, dropwhile  


if version_info[0] >= 3:
    from functools import reduce 
    imap = map  
else:
    from itertools import imap 
    
# ============================================================================ #
# DECORATORS                                                                   # 

class infix(object):
    """
    Given some function of prototype (functor, sequence) where the functor has a
    prototype of (x, y) such that y is the continuation value and x the item in
    the sequence, allow it to be accessed as an infixed operator.

    Eg:
    def function(functor, sequence):
        for i, qfunc in enumerate(sequence):
            yield qfunc(functor(i))

    Takes a sequence of functors and applies the index of the functor to the
    passed in functor operation, which is the input to the sequence functor.

    This can be called as such:

    function | rfold1 | function_sequence 
    
    """
    def __init__(self, functor):
        self.functor = functor 

    def __ror__(self, lhs):
        return infix(lambda x, self = self, lhs = lhs: self.functor(lhs, x))

    def __or__(self, rhs):
        return self.functor(rhs)

    def __rlshift__(self, lhs):
        return self.__ror__(lhs)

    def __rshift__(self, rhs):
        return self.__or__(rhs)

    def __call__(self, *args):
        return self.functor(*args)

# ============================================================================ #
# LIST UTILITIES                                                               # 

@infix 
def curry(functor, k, *args):
    """
    def test(*args):
        return args 
    
    composition = test | curry | 0 \
                       | curry | 1 \
                       | curry | 2 \
                       | curry | 3  

    print(composition(4))
    >>> [0, 1, 2, 3, 4] 
    """
    def curried(*args, **kwargs):
        return functor(*((k,) + args), **kwargs)

    return curried 

# ---------------------------------------------------------------------------- #
# predicate loops                                                              #

def until(condition, operation, value):
    """
    until(lambda x: x < .05, lambda y: y / 2, 100)
    >>> .025 (or some rounding error) 
    """
    while not condition(value):
        value = operation(value)

    return value

@infix  
def take_while(predicate, seq):
    """
    """
    return takewhile(predicate, seq)

@infix 
def drop_while(predicate, seq):
    """
    """
    return dropwhile(predicate, seq)

# ---------------------------------------------------------------------------- #
# list splitting                                                               #

@infix 
def span(predicate, seq):
    """
    span(lambda x: x < 3, [0, 1, 2, 3, 4, 5])
    >>> ([0, 1, 2, 3], [4, 5])
    """
    for k, r in enumerate(seq):
        if not predicate(r):
            return (seq[:k], seq[k:])
        
    return ([], seq) 


@infix 
def hbreak(predicate, seq):
    """
    hbreak(lambda x: x > 3, [0, 1, 2, 3, 4, 5])
    >>> ([0, 1, 2, 3], [4, 5])
    """
    for k, r in enumerate(seq):
        if predicate(r):
            return (seq[:k], seq[k:])
        
    return ([], seq) 

# ---------------------------------------------------------------------------- #
# generators                                                                   #

@infix 
def unfoldr(operation, seed):
    """
    list(((lambda x, y: x + y) | unfoldr | 1) | take 5)
    >>> [1, 1, 2, 3, 5]
    """
    curr = seed 

    while True:
        yield seed 
        
        inter = operation(seed, curr)
        seed  = curr 
        curr  = inter 

# ---------------------------------------------------------------------------- #
# interrogators                                                                #

@infix 
def hany(predicate, seq):
    """
    hany(lambda x: x % 2 != 0, [0, 2, 4, 5])
    >>> True 
    """
    for s in seq:
        if predicate(s): 
            return True 

    return False 

@infix
def hall(predicate, seq):
    """
    hall(lambda x: x % 2 != 0, [0, 2, 4, 5])
    >>> False 
    """
    for s in seq:
        if not predicate(s):
            return False 
    
    return True 

# ---------------------------------------------------------------------------- #
# accessors                                                                    #

def head(seq):
    """
    head([0, 1, 2])
    >>> 0
    """
    return seq[0]

def last(seq):
    """
    head([0, 1, 2])
    >>> 2 
    """
    return seq[-1]

def tail(seq):
    """
    head([0, 1, 2])
    >>> [1, 2] 
    """
    return seq[1:]

def init(seq):
    """
    head([0, 1, 2])
    >>> [0, 1] 
    """
    return seq[:-1]

@infix 
def take(seq, count):
    """
    list(range(100) | take | 2)
    >>> [0, 1]
    """
    for i, enum, in enumerate(seq):
        yield enum  

        if i == count - 1:
            break 

# ---------------------------------------------------------------------------- #
# destructors                                                                  #

def foldl(operation, sequence, default):
    return reduce(operation, sequence, default)

@infix 
def foldl1(operation, sequence):
    return reduce(operation, sequence)

def foldr(operation, sequence, default):
    return iter(foldl(operation, reversed(sequence), default))

@infix 
def foldr1(operation, sequence):
    return iter(foldl1(operation, reversed(list(sequence))))

def map_accuml(operation, mapper, seq):
    return operation | foldl1 | imap(mapper, seq)

def map_accumr(operation, mapper, seq):
    return operation | foldr1 | imap(mapper, seq)


# ---------------------------------------------------------------------------- #
# translators                                                                  #

@infix
def scanl1(operation, sequence):
    value = None

    for k, elem in enumerate(sequence):
        if 0 == k:
            value = elem 
        else:
            value = operation(value, elem)
        
        yield value 

def scanl(operation, sequence, default):
    for elem in sequence:
        default += operation(default, elem)
        yield default 


# ---------------------------------------------------------------------------- #
# injectors                                                                    #

def intersperse(elem, seq):
    ret = [None] * (2 * len(seq) - 1)

    ret[0::2] = seq 
    ret[1::2] = [elem for _ in range(len(seq) - 1)]

    return chain(*ret) 

def intercalate(seq, seqseq):
    return intersperse(seq, seqseq)
 

# ---------------------------------------------------------------------------- #
# sequencers                                                                   #

def transpose(seq):
    ret = [[] for _ in range(max(map(len, seq)))] 
    
    for i, r in enumerate(ret): 
        for s in seq:
            if len(s) <= i:
                continue 
            else:
                print(s, i)
                r.append(s[i])
    
    return ret 

def subsequences(seq):
    for i in range(len(seq) + 1):
        for k in combinations(seq, i):
            yield list(k) 


# ============================================================================ #
# SEQUENCE ASSERTION                                                           # 

class Monotonic(object):
    """ Iterate on a list executing `operation` on each element, raise a
    RuntimeError if at any point the sequence is proven non-monotonic.
    """
    def __init__(self, seq, operation = lambda x: x):
        self.sequence    = seq 
        self.disposition = None 
        self.operation   = operation 

    def __iter__(self):
        step_n0 = None

        for elem in imap(self.operation, self.sequence):
            if self.disposition is None:
                if step_n0 is None:
                    step_n0 = elem 
                else:
                    if elem != step_n0:
                        self.disposition = (gt, lt)[elem > step_n0]

            if self.disposition:
                if self.disposition(elem, step_n0):
                    raise RuntimeError('SEQUENCE NOT MONOTONIC')
                else:
                    step_n0 = elem  

            yield elem 
            

# ============================================================================ #
# FUNCTIONAL COMPOSITION                                                       # 

def fold(monoid, args):
    return monoid.__fold__(args)

class Monoid(object):
    """
    """

    def __init__(self, operation, identity, null):
        self.operation = lambda x, y: operation(x, identity(y))
        self.identity  = identity
        self.null      = null 
    
    def __fold__(self, args):
        if 0 == len(args):
            return self.null 
        else:
            return foldl(self.operation, args, self.null)

    def __add__(self, m):
        """ 
        :param m:   
        :type  m:   Monoid 

        Given some monoid added to self, compose the functions lazily such that
        when it folds, it folds left in the composition calling curried down
        higherarchy from right monoid to left.

        fold(m_0 + m_1, seq)
        
        for s in seq: fold(m_1(m_0(s))

        """
        assert isinstance(m, Monoid)   
        assert self.null                      == m.null 
        assert type(self.identity(self.null)) == type(sm.identity(m.null))

        l = lambda x, y: self.operation(m.operation(x, m.identity(y))
                                      , self.identity(y))

        return Monoid(l, self.identity, self.null)

    def __call__(self, seq):
        return fold(self, seq) 


#class do(object):
#    def __init__(self, functor):
#        self.functor = functor 
#        
#    def __call__(self, *args, **kwargs):
#        
#        def determine():
#            iterator = self.functor(*args, **kwargs)
#            pass            
#
#
#class Monad(object):
#    def bind(self, functor):
#        raise NotImplementedError('Monad needs a bind operator defined')
#    
#    def __irshift__(self, functor):
#        return self.bind(functor)
#    
#class Failable(Monad):
#    def __init__(self, value, status):
#        self.value  = value 
#        self.status = status 
#
#    def bind(self, functor):
#        if self.status:
#            return functor(self.value)
#        else:
#            return self 


# ============================================================================ #
# FUNCTION ACCESSORS                                                           # 

class MatchObj(object):
    """
    The MatchObj class is used to create a decorator class that when called
    attempts to match the arguments to the appropriate function signature and
    forward the call.

    match_object = MatchObj()
    
    @match_object.pattern 
    def function0(arg0 = int, arg1 = []):
        print('HERE-0')
    
    @match_object.pattern 
    def function1(arg0 = {} , arg1 = []):
        print('HERE-1')

    
    match_object({'keys': 'value'}, [0, 1, 2])
    >>> HERE-1
    
    Preliminary estimates show in "best case" scenario, matching is 4x slower,
    in worst case it is 6x slower.  So if you have a tight recursion which needs
    to be fast, matching here is not recommended.
    """
    def __init__(self):
        self.__lookup = []
        self.__engine = None 

    def pattern(self, functor):
        """ Use this function as the object decorator over functions that should
        be pattern matched in this group.
        """
        args = [ k if isinstance(k, type) else type(k) for k in \
                 getargspec(functor).defaults ]
        
        self.__lookup.append((functor, args))
    
    def __match_len(self, args):
        """ Matches the table by argument length only since it has been proven
        in `compile` that all matchable function signature have unique lengths 
        """
        return self.__table[len(args)](*args)        

    def __index(self, args):
        """ At least one index is unique amongst all arguments, this function
        does a table look up of the type at the argument index.
        """
        return self.__table[type(args[self.__index])](*args)
    
    def __len_arg(self, args):
        """ There is neither a unique argument length or a unique index of
        arguments amongst signatures.  However every function by definition must
        have either a different number of arguments, or, one argument which is
        unique from function signature of the same length.  This matches that case.
        """
        lut  = self.__table[len(args)]
        return self.__table[len(args)][1][type(args[lut[0]])](*args)

    def compile(self):
        """ Given a set of function signature, there may be more optimized ways
        of proving a match than the last resort method.  Find the most efficient
        way and bind it to the private __engine reference along with any
        necessary metadata.
        """
        # check if every function has different number of variables 
        match = { len(k[1]) for k in self.__lookup }
        
        if len(match) == len(self.__lookup):
            self.__engine = self.__match_len 
            self.__table  = {len(k[1]):k[0] for k in self.__lookup}
            
            return 
        
        # check if there is atleast one index position which every function
        # differs 
        for i in range(max(map(lambda x: len(x[1]), self.__lookup))):
            types = [None] * len(self.__lookup)
            elems = set([]) 

            for j, entry in enumerate(self.__lookup):
                if len(entry[1]) <= i:
                    continue
                else:
                    types[j]    = entry[1][i]
                    elems.add(str(entry[1][i]))
            
            # Index => i must be unique 
            if len(elems) == len(types):
                self.__engine = self.__index 
                self.__table  = {k[1][i]:k[0] for k in self.__lookup}
                self.__index  = i
                
                return  
            
        
        # If the patterns are valid, there must be at least one unique member of
        # each argument prototype of length k 
        self.__engine = self.__len_arg 
        self.__table  = {} 
        
        for i in set(map(lambda x: len(x[1]), self.__lookup)):
            
            matching = list(filter(lambda x: len(x[1]) == i, self.__lookup))
            self.__table[i] = [None, {}] 
            
            for j in range(i):
                types = [None] * i 
                elems = set([])

                for k, m in enumerate(matching):
                    types[k]  = m[1]
                    elems.add(str(m[1]))

                if len(types) == len(elems):
                    self.__table[i][0] = j
                    self.__table[i][1] = {q[1][j]:q[0] for q in matching}
                    

        # assert the table is sane 
        count = 0

        for _, items in self.__table.items():
            count += len(items[1]) 
        
        if count != len(self.__lookup):
            raise RuntimeError('INVALID PATTERN MATCHING TABLE')
        else:
            del(self.__lookup)


    def __call__(self, *args):
        """ Passes the arguments along to the function matching engine chosen in
        compile.
        """
        return self.__engine(args) 
    


if __name__ == '__main__':
    a = (lambda x, y: x * y) | curry | 3 
    b = (lambda x, y: x + y) | curry | 'Hello '
    assert a(4)       == 12 
    assert b('World') == 'Hello World'
    
    assert 1 == until(lambda x: x <  2, lambda x: x / 2, 1024)
    assert 2 == until(lambda x: x <= 2, lambda x: x / 2, 1024)

    assert 0      == head([0, 1, 2])
    assert [1, 2] == tail([0, 1, 2])
    assert 2      == last([0, 1, 2])
    assert [0, 1] == init([0, 1, 2])

    assert      (lambda x: x % 2 == 0)  | hany  | [1, 3, 5, 2]
    assert not ((lambda x: x % 2 == 0) << hany >> [1, 3, 5, 7])
    assert      (lambda x: x % 2 == 0)  | hall  | [0, 2, 4, 8]
    assert not ((lambda x: x % 2 == 0) << hall >> [0, 2, 4, 7])
    
    assert 30 == (lambda x, y: x + y) | foldr1 | [2, 4, 8, 16] 
    assert 30 == (lambda x, y: x + y) | foldl1 | [2, 4, 8, 16] 
   
    assert [1, 2, 3] == list((lambda x, y: x + y) | scanl1 | [1, 1, 1])

    
    assert [2, 4] == (lambda x: x % 2 == 0) | drop_while | [1, 3, 2, 4]
    assert [1, 3] == (lambda x: x % 2 != 0) | take_while | [1, 3, 2, 4]

    assert ([0, 1], [2, 3]) == (lambda x: x < 2) | span   | [0, 1, 2, 3]
    assert ([0, 1], [2, 3]) == (lambda x: x > 1) | hbreak | [0, 1, 2, 3]
    
    summer    = Monoid(lambda x, y: x + y              , lambda x: x, 0)
    collector = Monoid(lambda x, y: x + fold(summer, y), lambda x: x, 0) 
    
    mq = Monoid(lambda x, y: x + y, lambda x: x, 0)
    mr = Monoid(lambda x, y: x * y, lambda x: x, 0)
    
    assert fold(mq + mr, [1, 2]) != fold(mr + mq, [1, 2])
    assert 12 == (fold(summer   , [3, 4, 5]))
    assert 21 == (fold(collector, [ [1, 2, 3], [4, 5, 6] ]))
    
    for elem in Monotonic([0, 1, 2, 3, 3, 4]):
        pass  
    
    try:
        for elem in Monotonic([0, 1, 2, 3, 2, 4]):
            pass 
    except RuntimeError as err:
        pass      
    else:
        assert 'monotonic' == False 
     
    for elem in Monotonic(reversed([0, 1, 2, 3, 3, 4])):
        pass  
    
    try:
        for elem in Monotonic(reversed([0, 1, 2, 3, 2, 4])):
            pass 
    except RuntimeError as err:
        pass      
    else:
        assert 'monotonic' == False 
   
    for elem in Monotonic([0, 1, 2, 3, 3, 4], lambda x: x ** 2):
        pass  
    
    #print(''.join(intersperse(' ', ['words', 'here'])))
    #print(list(intercalate([0, 0], [[1, 1, 1], [2, 2]])))
    #print(transpose([[1, 2, 3], [4, 5, 6], [7, 8, 9]]))
    #print(transpose(['hey', 'there', 'guys']))
    #
    #print(list(subsequences('abc')))

    #print(map_accuml(lambda x, y: x + y, str, [0, 1, 2]))
    ##print(map_accumr(lambda x, y: x + y, str, [0, 1, 2]))

    #print(
    #        list(((lambda x, y: x + y) | unfoldr | 1) | take | 8)
    #        )

    #print(list((lambda x, y: x + y) | unfoldr | 1))  
    #m = Monad()
    #m >>= lambda x: x