working abstract class implementation

.gitignore

@@ -1,2 +1,3 @@
 **/.venv/
-**/__pycache__
+**/__pycache__
+**/.DS_Store

examples/caltech_courses/.gitignore

@@ -1 +1,2 @@
 .venv/
+pages/

examples/caltech_courses/parse.py

@@ -6,8 +6,8 @@
 from os.path import join as path_join
 from glob import glob
 
-from free_flow import ff, rff, T, dangerous_eval_attr as dea
-from operator import methodcaller as mc, itemgetter as ig, attrgetter as ag
+from free_flow import ff, rff, T, dea, mc, ig, ag
+# from operator import methodcaller as mc, itemgetter as ig, attrgetter as ag
 
 with open('./options.xml', 'r') as rf:
     s = rf.read()
@@ -39,7 +39,11 @@ def inspect(x):
            #      mc('select', '.course-description2__title'),
            #      )
            #),
-           rff(T(   [ mc('select', '.course-description2__title'), dea('[0].text.strip()') ],  ag('text')   ) )
+           rff(T(
+                    [ mc('select', '.course-description2__title'), dea('[0].text.strip()') ],
+                    ag('text')
+            ))
+
            #rff(T(   [ mc('select', '.course-description2__title'), ig(0), ag('text'), mc('strip') ],  ag('text')   ) )
        ]
 
@@ -55,9 +59,13 @@ def read_to_soup(path):
     )(read_to_soup, *fs)
 
 x = data(courses)
-print(x)
+
+# x = ff(glob('./pages/*.html'))(read_to_soup, *courses)
+# print(x)
 flat = [a for y in x for a in y]
 
+
+print(flat[:10])
 #from json import dump
 #with open('all_courses.json', 'w') as wf:
 #    dump(flat, wf)

free_flow/__init__.py

@@ -1 +1 @@
-from .lib import ff, rff, T, dangerous_eval_attr
+from .segments import ff, rff, Tee as T, dea, mc, ig, ag

free_flow/lib.py

@@ -33,54 +33,7 @@ def ret(x):
     return ret
 
 
-# def compose(*funcs: List[Callable]):
-#     def composition(x):
-#         for f in funcs:
-#             x = f(x)
-#         return x
-#     return composition
-
-
-# class T(Segment):
-#     def __init__(self, *funcs_lists: Iterable[Callable or Iterable[Callable]]):
-#         self.funcs = [compose(*fl) if isinstance(fl, Iterable) else fl for fl in funcs_lists]
-
-#     def __call__(self, x):
-#         if self.eager: return [f(x) for f in self.funcs]
-#         else: return map(lambda f: f(x), self.funcs)
-
-def getname(obj):
-    if isinstance(obj, Iterable):
-        return ' ⮕ '.join(getname(o) for o in obj)
-    if not hasattr(obj, '__qualname__'):
-        return obj.__name__
-    if callable(obj) and obj.__qualname__ == '<lambda>':
-        # console.print("hewwwwwwwwwwwwwwwwwwww", obj, style="blue")
-        # console.print(getsource(obj))
-        # console.print(type(getsource(obj)))
-        # console.print("\n\n\n")
-        # return 'λ(' + getsource(obj) + ')'
-        return 'λ'
-    return obj.__qualname__
-
-def ff(x_arr, eager=ff_default_eager, test_single=False):
-    if test_single: x_arr = x_arr[:1]
-    def gen(*funcs: List[Callable]):
-        for x in x_arr:
-            yield safe_compose(*funcs)(x)
-    def gen_eager(*funcs: List[Callable]):
-        # for f in funcs:
-        #     if isinstance(f, Segment):
-        #         f.eager = True
-        return [safe_compose(*funcs)(x) for x in x_arr]
-    return gen_eager if eager else gen
-
-# same as ff but take the functions as curry first, before the data
-def rff(*funcs: List[Callable], eager=ff_default_eager, test_single=False):
-    def inner(x_arr):
-        return ff(x_arr, eager, test_single)(*funcs)
-    return inner
-
+from segments import ff
 # TODO: print the graph! https://github.com/pydot/pydot
 # TODO: make a flatten utility? and a *apply utility? (eg. something turns arr into arr[arr], then i want to apply smt to each thing of the inner
 

free_flow/segments.py

@@ -1,9 +1,20 @@
 from abc import ABC, abstractmethod, abstractproperty
 from inspect import signature, getsource
 from typing import Any, Iterable, Callable, List
+from types import BuiltinFunctionType, BuiltinMethodType
 from rich.console import Console
 console = Console()
 
+class FlowError(Exception):
+    """Raised when something fails to flow through the pipe."""
+    def __init__(self, inner, fn, value):
+        self.inner = inner
+        self.fn = fn
+        self.value = value
+
+    def __repr__(self):
+        return f"[red][bold]{type(self.inner).__name__}[/bold]: {self.inner}[/red] on [green]{self.value}[/green] → [blue]{repr(self.fn)}[/blue]"
+
 class Segment(ABC):
     def __init__(self):
         pass
@@ -36,7 +47,10 @@ def __init__(self, fn):
 
     def __call__(self, x): return self.fn(x)
 
-    def __repr__(self): return getsource(self.fn).strip()
+    def __repr__(self):
+        if isinstance(self.fn, BuiltinFunctionType) or isinstance(self.fn, BuiltinMethodType):
+            return self.fn.__name__
+        return getsource(self.fn).strip()
 
     @property
     def input_types(self):
@@ -75,14 +89,17 @@ def __call__(self, x): return itemgetter(self.expr)(x)
     def __repr__(self): return f'[{self.expr}]'
 
 class MethodCaller(Segment):
-    def __init__(self, expr): self.expr = expr
-    def __call__(self, x): return methodcaller(self.expr)(x)
-    def __repr__(self): return f'({self.expr})'
+    def __init__(self, method, *args, **kwargs):
+        self.method = method
+        self.args = args
+        self.kwargs = kwargs
+    def __call__(self, x): return methodcaller(self.method, *self.args, **self.kwargs)(x)
+    def __repr__(self): return f".{self.method}({', '.join(list(self.args) + [f'{k}={v}' for k, v in self.kwargs.items()] )})"
 
 class AttrGetter(Segment):
-    def __init__(self, expr): self.expr = expr
-    def __call__(self, x): return attrgetter(self.expr)(x)
-    def __repr__(self): return f'.{self.expr}'
+    def __init__(self, attr): self.attr = attr
+    def __call__(self, x): return attrgetter(self.attr)(x)
+    def __repr__(self): return f'.{self.attr}'
 
 dea = DangerousEvalAttr
 mc = MethodCaller
@@ -103,57 +120,90 @@ def __repr__(self): return '>' + repr(self.seg)
 si = SpreadInto
 
 class Compose(Segment):
-    def __init__(self, segs: List[Segment]): self.segs = segs
+    def __init__(self, *segs: List[Segment]): self.segs = segs
     def __call__(self, x):
         for f in self.segs:
             try:
                 console.print(f"[green]{type(x).__name__}[/green] ➡️ {repr(f)}")
                 x = f(x)    # optm: add vectorizability
             except Exception as e:
-                # raise ValueError(f"got error {e} when applying {repr(f)} to {x}")    # todo: actually helpful errors that tell you what went wrong and what the types/values involved were
-                console.print(f"[red][bold]{type(e).__name__}[/bold]: {e}[/red] on [green]{x}[/green] → [blue]{repr(f)}[/blue]")
-                raise e
+                raise FlowError(e, f, x)
         return x
     def __repr__(self): return ' ⮕ '.join(repr(s) for s in self.segs)
 
 class Tee(Segment):
-    def __init__(self, funcs_lists: List[Segment]):
-        self.funcs = [Compose(*fl) if isinstance(fl, Iterable) else fl for fl in funcs_lists]
+    def __init__(self, *segs: Iterable[Segment]):
+        self.funcs = [ingestor(fl) for fl in segs]
     def __call__(self, x):
         return [f(x) for f in self.funcs]
     def __repr__(self):
         return '{ ' + '; '.join(repr(fl) for fl in self.funcs) + ' }'
 
 class LazyTee(Tee):
-    def __init__(self, funcs_list: List[Segment]):
-        return super().__init__(self, funcs_list)
+    def __init__(self, *segs: Iterable[Segment]):
+        return super().__init__(self, segs)
     def __call__(self, x):
         return map(lambda f: f(x), self.funcs)
 
 def ingestor(obj: Segment or set or Iterable or Callable):
     if isinstance(obj, Segment):
         return obj
     if isinstance(obj, set):
-        return Tee([ingestor(f) for f in obj])
+        return Tee(*[ingestor(f) for f in obj])
     if isinstance(obj, Iterable):
-        return Compose([ingestor(f) for f in obj])
+        return Compose(*[ingestor(f) for f in obj])
     if isinstance(obj, Callable):
         return FunctionAdapter(obj)
     else:
         raise NotImplementedError(f"Could not ingest object of type {type(obj)}")
 
+ff_default_eager = True
+class Pipeline(Segment):
+    def __init__(self, *funcs: List[Callable], eager=ff_default_eager, test_single=False):
+        self.pipe = ingestor(funcs)
+        self.eager = eager
+
+    def __call__(self, x_arr):
+        if self.eager:
+            for x in x_arr:
+                yield self.pipe(x)
+        else:
+            return [self.pipe(x) for x in x_arr]
+    def __repr__(self):
+        return '...' + repr(self.pipe)
+rff = Pipeline
+
+def ff(x_arr, eager=ff_default_eager, test_single=False):
+    if test_single: x_arr = x_arr[:1]
+    def ret(*funcs):
+        try:
+            pipe = ingestor(funcs)
+            return [pipe(x) for x in x_arr] # todo: non-eager version
+        except FlowError as fe:
+            console.print(fe)
+    return ret
+
+
 if __name__ == '__main__':
+
+
+    y = ff("hello")
+    print(y(ord))
+
+
+
     # dea = DangerousEvalAttr(".strip().split('.')")
     # print(dea('hello.world      .emacs.    '))
     # print(dea)
 
     # op = MethodCaller('cow')
     # # print(op({ 'pig': 'oink', 'cow': 'moo' }))
 
-    pipe = ingestor([
-        dea(".strip().split('.')"),
-        # vec(lambda s: s.upper())
-        lambda s: s.upper()
-        ])
-    print(pipe)
-    print(pipe('hello.world      .emacs.    '))
+
+    # pipe = ingestor([
+    #     dea(".strip().split('.')"),
+    #     # vec(lambda s: s.upper())
+    #     lambda s: s.upper()
+    #     ])
+    # print(pipe)
+    # print(pipe('hello.world      .emacs.    '))