rebuild and retest

Author: JohnMount

build/lib/data_algebra/arrow.py

@@ -2,9 +2,10 @@
 
 import data_algebra.data_ops
 import data_algebra.flow_text
+from data_algebra.shift_pipe_action import ShiftPipeAction
 
 
-class Arrow(abc.ABC):
+class Arrow(ShiftPipeAction):
     """
     Arrow from category theory: see Steve Awody,
     "Category Theory, 2nd Edition", Oxford Univ. Press, 2010 pg. 4.
@@ -13,7 +14,7 @@ class Arrow(abc.ABC):
     """
 
     def __init__(self):
-        pass
+        ShiftPipeAction.__init__(self)
 
     @abc.abstractmethod
     def dom(self):
@@ -23,28 +24,16 @@ def dom(self):
     def cod(self):
         """return co-domain, object at head of arrow"""
 
-    @abc.abstractmethod
-    def apply_to(self, b):
-        """apply_to b, compose arrows (right to left)"""
-
     # noinspection PyPep8Naming
     @abc.abstractmethod
-    def act_on(self, X):
-        """act on X, must associate with composition"""
+    def act_on(self, b):
+        """act on b, must associate with composition"""
 
     # noinspection PyPep8Naming
     def transform(self, X):
         """transform X, may or may not associate with composition"""
         return self.act_on(X)
 
-    def __rshift__(self, other):  # override self >> other
-        return other.apply_to(self)
-
-    def __rrshift__(self, other):  # override other >> self
-        if isinstance(other, Arrow):
-            return self.apply_to(other)
-        return self.act_on(other)
-
 
 class DataOpArrow(Arrow):
     """
@@ -79,39 +68,44 @@ def get_feature_names(self):
         cp = self.outgoing_columns.copy()
         return cp
 
-    def apply_to(self, b):
-        """replace self input table with b"""
+    def act_on(self, b, *, correct_ordered_first_call: bool = False):
+        """
+        Apply self onto b.
+        
+        :param b: item to act on, or item that has been sent to self.
+        :param correct_ordered_first_call: if True indicates this call is from __rshift__ or __rrshift__ and not the fallback paths.
+        """
+        assert isinstance(correct_ordered_first_call, bool)
         if isinstance(b, data_algebra.data_ops.ViewRepresentation):
             b = DataOpArrow(b)
-        assert isinstance(b, DataOpArrow)
-        # check categorical arrow composition conditions
-        missing = set(self.incoming_columns) - set(b.outgoing_columns)
-        if len(missing) > 0:
-            raise ValueError("missing required columns: " + str(missing))
-        excess = set(b.outgoing_columns) - set(self.incoming_columns)
-        if len(excess) > 0:
-            raise ValueError("extra incoming columns: " + str(excess))
-        new_pipeline = self.pipeline.replace_leaves({self.free_table_key: b.pipeline})
-        new_pipeline.get_tables()  # check tables are compatible
-        res = DataOpArrow(
-            pipeline=new_pipeline,
-            free_table_key=b.free_table_key,
-        )
-        return res
-
-    # noinspection PyPep8Naming
-    def act_on(self, X):
+        if isinstance(b, DataOpArrow):
+            # check categorical arrow composition conditions
+            missing = set(self.incoming_columns) - set(b.outgoing_columns)
+            if len(missing) > 0:
+                raise ValueError("missing required columns: " + str(missing))
+            excess = set(b.outgoing_columns) - set(self.incoming_columns)
+            if len(excess) > 0:
+                raise ValueError("extra incoming columns: " + str(excess))
+            new_pipeline = self.pipeline.replace_leaves({self.free_table_key: b.pipeline})
+            new_pipeline.get_tables()  # check tables are compatible
+            res = DataOpArrow(
+                pipeline=new_pipeline,
+                free_table_key=b.free_table_key,
+            )
+            return res
+        if correct_ordered_first_call and isinstance(b, ShiftPipeAction):
+            return b.act_on(self, correct_ordered_first_call=False)  # fall back
         # assume a pandas.DataFrame compatible object
         # noinspection PyUnresolvedReferences
-        cols = set(X.columns)
+        cols = set(b.columns)
         missing = set(self.incoming_columns) - cols
         if len(missing) > 0:
             raise ValueError("missing required columns: " + str(missing))
         excess = cols - set(self.incoming_columns)
         assert len(excess) == 0
         if len(excess) > 0:
-            X = X[self.incoming_columns]
-        return self.pipeline.act_on(X)
+            b = b[self.incoming_columns]
+        return self.pipeline.act_on(b)
 
     def dom(self):
         return DataOpArrow(

build/lib/data_algebra/cdata.py

@@ -10,6 +10,7 @@
 import data_algebra.data_model
 import data_algebra.util
 import data_algebra.data_ops
+from data_algebra.shift_pipe_action import ShiftPipeAction
 
 
 def _str_list_to_html(lst: Iterable[str]) -> str:
@@ -257,7 +258,7 @@ def map_from_rows(self):
         return RecordMap(blocks_out=self, strict=self.strict)
 
 
-class RecordMap:
+class RecordMap(ShiftPipeAction):
     """
     Class for specifying general record to record transforms.
     """
@@ -280,6 +281,7 @@ def __init__(
         :param blocks_out: outgoing record specification, None for row-records.
         :param strict: if True insist block be strict, and in and out blocks agree on row-form columns.∂
         """
+        ShiftPipeAction.__init__(self)
         assert isinstance(strict, bool)
         self.strict = strict
         if blocks_in is not None:
@@ -426,10 +428,6 @@ def transform(
                 blocks_out=self.blocks_out,
             )
         return X
-    
-    def act_on(self, X):
-        assert isinstance(X, data_algebra.data_ops.ViewRepresentation)
-        return X.convert_records(self)
 
     def compose(self, other):
         """
@@ -534,12 +532,17 @@ def inverse(self):
         """
         assert self.strict
         return RecordMap(blocks_in=self.blocks_out, blocks_out=self.blocks_in, strict=True)
-
-    def __rshift__(self, other):  # override self >> other
-        return self.act_on(other)
-
-    def __rrshift__(self, other):  # override other >> self
-        return self.transform(other)
+    
+    def act_on(self, b, *, correct_ordered_first_call: bool = False):
+        assert isinstance(correct_ordered_first_call, bool)
+        if isinstance(b, RecordMap):
+            self.compose(b)
+        if isinstance(b, data_algebra.data_ops.ViewRepresentation):
+            return b.convert_records(self)
+        if correct_ordered_first_call and isinstance(b, ShiftPipeAction):
+            return b.act_on(self, correct_ordered_first_call=False)  # fall back to peer's action
+        # assume table like
+        return self.transform(b)
 
     def fmt(self) -> str:
         """Format for informal presentation."""

build/lib/data_algebra/data_ops.py

@@ -24,6 +24,7 @@
     ordered_union,
 )
 import data_algebra.util
+from data_algebra.shift_pipe_action import ShiftPipeAction
 
 
 _have_black = False
@@ -411,36 +412,38 @@ def __ne__(self, other):
 
     # composition
 
-    # noinspection PyPep8Naming
-    def act_on(self, X, *, data_model=None):
+    def act_on(self, b, *, correct_ordered_first_call: bool = False):
         """
-        apply self to X, must associate with composition
+        apply self to b, must associate with composition
         Operator is strict about column names.
 
-        :param X: input data frame
-        :param data_model implementation to use
-        :return: transformed result
+        :param b: input data frame
+        :param correct_ordered_first_call: indicate not on fallback path
+        :return: transformed or composed result
         """
+        assert isinstance(correct_ordered_first_call, bool)
         tables = self.get_tables()
-        if isinstance(X, ViewRepresentation):
+        if isinstance(b, ViewRepresentation):
             # insert to only table or if more than one, table with matching key
             if len(tables) == 1:
                 key = list(tables.keys())[0]
             else:
-                key = X.key
+                key = b.key
             assert isinstance(key, str)
             old = tables[key]
-            assert set(X.column_names) == set(old.column_names)  # this is defending associativity of composition against table narrowing
-            return self.replace_leaves({key: X})
+            assert set(b.column_names) == set(old.column_names)  # this is defending associativity of composition against table narrowing
+            return self.replace_leaves({key: b})
+        # see if b is ShiftPipeAction, so it can handle the mapping (using fact data is not a ShiftPipeAction instance)
+        if correct_ordered_first_call and isinstance(b, ShiftPipeAction):
+            return b.act_on(self, correct_ordered_first_call=False)
         # assume a table
         assert len(tables) == 1
         key = list(tables.keys())[0]
         assert isinstance(key, str)
         old = tables[key]
-        assert set(X.columns) == set(old.column_names)
+        assert set(b.columns) == set(old.column_names)
         return self.transform(
-            X=X,
-            data_model=data_model,
+            b,
             strict=True,
         )
 

build/lib/data_algebra/data_ops_types.py

@@ -8,6 +8,7 @@
 import data_algebra.expr_rep
 import data_algebra.cdata
 import data_algebra.OrderedSet
+from data_algebra.shift_pipe_action import ShiftPipeAction
 
 
 class MethodUse(NamedTuple):
@@ -19,12 +20,13 @@ class MethodUse(NamedTuple):
     is_ordered: bool = False
 
 
-class OperatorPlatform(abc.ABC):
+class OperatorPlatform(ShiftPipeAction):
     """Abstract class representing ability to apply data_algebra operations."""
 
     node_name: str
 
     def __init__(self, *, node_name: str):
+        ShiftPipeAction.__init__(self)
         assert isinstance(node_name, str)
         self.node_name = node_name
 
@@ -73,24 +75,6 @@ def replace_leaves(self, replacement_map: Dict[str, Any]):
         :return: new operator DAG
         """
 
-    # noinspection PyPep8Naming
-    @abc.abstractmethod
-    def act_on(self, X, *, data_model=None):
-        """
-        apply self to X, must associate with composition
-        Operator is strict about column names.
-
-        :param X: input data frame
-        :param data_model implementation to use
-        :return: transformed result
-        """
-
-    def __rshift__(self, other):  # override self >> other
-        return other.act_on(self)
-
-    def __rrshift__(self, other):  # override other >> self
-        return self.act_on(other)
-
     # imitate a method
     def use(self, user_function, *args, **kwargs):
         """

build/lib/data_algebra/shift_pipe_action.py

@@ -0,0 +1,38 @@
+
+import abc
+
+
+class ShiftPipeAction(abc.ABC):
+    """
+    Class representing mapping a >> b to b.act_on(a). 
+    This is read as "sending a to b".
+    """
+    def __init__(self) -> None:
+        pass
+
+    @abc.abstractmethod
+    def act_on(self, b, *, correct_ordered_first_call: bool = False):
+        """
+        Apply self onto b.
+        
+        :param b: item to act on, or item that has been sent to self.
+        :param correct_ordered_first_call: if True indicates this call is from __rshift__ or __rrshift__ and not the fallback paths.
+        """
+
+    def __rshift__(self, b):  # override self >> b
+        """
+        Delegate self >> b to b.act_on(self) b is a ShiftPipeAction instance, else call self.act_on(b)
+        This is read as "sending self to b".
+        """
+        if isinstance(b, ShiftPipeAction):
+            # this is the expected path
+            return b.act_on(self, correct_ordered_first_call=True)
+        # fall back to our action
+        return self.act_on(b, correct_ordered_first_call=False)
+
+    def __rrshift__(self, b):  # override b >> self
+        """
+        Delegate b >> self to self.act_on(b).
+        This is read as sending b to self.
+        """
+        return self.act_on(b, correct_ordered_first_call=True)